1 files changed, 419 insertions, 0 deletions
diff --git a/lib/sqlalchemy/pool/impl.py b/lib/sqlalchemy/pool/impl.py
new file mode 100644
index 000000000..c45661895
--- /dev/null
+++ b/lib/sqlalchemy/pool/impl.py
@@ -0,0 +1,419 @@
+# sqlalchemy/pool.py
+# Copyright (C) 2005-2018 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+
+"""Pool implementation classes.
+
+"""
+
+import traceback
+import weakref
+
+from .base import Pool, _ConnectionRecord
+from .. import exc
+from .. import util
+from ..util import queue as sqla_queue
+from ..util import chop_traceback
+from ..util import threading
+
+
+class QueuePool(Pool):
+
+    """A :class:`.Pool` that imposes a limit on the number of open connections.
+
+    :class:`.QueuePool` is the default pooling implementation used for
+    all :class:`.Engine` objects, unless the SQLite dialect is in use.
+
+    """
+
+    def __init__(self, creator, pool_size=5, max_overflow=10, timeout=30,
+                 **kw):
+        r"""
+        Construct a QueuePool.
+
+        :param creator: a callable function that returns a DB-API
+          connection object, same as that of :paramref:`.Pool.creator`.
+
+        :param pool_size: The size of the pool to be maintained,
+          defaults to 5. This is the largest number of connections that
+          will be kept persistently in the pool. Note that the pool
+          begins with no connections; once this number of connections
+          is requested, that number of connections will remain.
+          ``pool_size`` can be set to 0 to indicate no size limit; to
+          disable pooling, use a :class:`~sqlalchemy.pool.NullPool`
+          instead.
+
+        :param max_overflow: The maximum overflow size of the
+          pool. When the number of checked-out connections reaches the
+          size set in pool_size, additional connections will be
+          returned up to this limit. When those additional connections
+          are returned to the pool, they are disconnected and
+          discarded. It follows then that the total number of
+          simultaneous connections the pool will allow is pool_size +
+          `max_overflow`, and the total number of "sleeping"
+          connections the pool will allow is pool_size. `max_overflow`
+          can be set to -1 to indicate no overflow limit; no limit
+          will be placed on the total number of concurrent
+          connections. Defaults to 10.
+
+        :param timeout: The number of seconds to wait before giving up
+          on returning a connection. Defaults to 30.
+
+        :param \**kw: Other keyword arguments including
+          :paramref:`.Pool.recycle`, :paramref:`.Pool.echo`,
+          :paramref:`.Pool.reset_on_return` and others are passed to the
+          :class:`.Pool` constructor.
+
+        """
+        Pool.__init__(self, creator, **kw)
+        self._pool = sqla_queue.Queue(pool_size)
+        self._overflow = 0 - pool_size
+        self._max_overflow = max_overflow
+        self._timeout = timeout
+        self._overflow_lock = threading.Lock()
+
+    def _do_return_conn(self, conn):
+        try:
+            self._pool.put(conn, False)
+        except sqla_queue.Full:
+            try:
+                conn.close()
+            finally:
+                self._dec_overflow()
+
+    def _do_get(self):
+        use_overflow = self._max_overflow > -1
+
+        try:
+            wait = use_overflow and self._overflow >= self._max_overflow
+            return self._pool.get(wait, self._timeout)
+        except sqla_queue.Empty:
+            # don't do things inside of "except Empty", because when we say
+            # we timed out or can't connect and raise, Python 3 tells
+            # people the real error is queue.Empty which it isn't.
+            pass
+        if use_overflow and self._overflow >= self._max_overflow:
+            if not wait:
+                return self._do_get()
+            else:
+                raise exc.TimeoutError(
+                    "QueuePool limit of size %d overflow %d reached, "
+                    "connection timed out, timeout %d" %
+                    (self.size(), self.overflow(), self._timeout), code="3o7r")
+
+        if self._inc_overflow():
+            try:
+                return self._create_connection()
+            except:
+                with util.safe_reraise():
+                    self._dec_overflow()
+        else:
+            return self._do_get()
+
+    def _inc_overflow(self):
+        if self._max_overflow == -1:
+            self._overflow += 1
+            return True
+        with self._overflow_lock:
+            if self._overflow < self._max_overflow:
+                self._overflow += 1
+                return True
+            else:
+                return False
+
+    def _dec_overflow(self):
+        if self._max_overflow == -1:
+            self._overflow -= 1
+            return True
+        with self._overflow_lock:
+            self._overflow -= 1
+            return True
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+        return self.__class__(self._creator, pool_size=self._pool.maxsize,
+                              max_overflow=self._max_overflow,
+                              timeout=self._timeout,
+                              recycle=self._recycle, echo=self.echo,
+                              logging_name=self._orig_logging_name,
+                              use_threadlocal=self._use_threadlocal,
+                              reset_on_return=self._reset_on_return,
+                              _dispatch=self.dispatch,
+                              dialect=self._dialect)
+
+    def dispose(self):
+        while True:
+            try:
+                conn = self._pool.get(False)
+                conn.close()
+            except sqla_queue.Empty:
+                break
+
+        self._overflow = 0 - self.size()
+        self.logger.info("Pool disposed. %s", self.status())
+
+    def status(self):
+        return "Pool size: %d  Connections in pool: %d "\
+            "Current Overflow: %d Current Checked out "\
+            "connections: %d" % (self.size(),
+                                 self.checkedin(),
+                                 self.overflow(),
+                                 self.checkedout())
+
+    def size(self):
+        return self._pool.maxsize
+
+    def checkedin(self):
+        return self._pool.qsize()
+
+    def overflow(self):
+        return self._overflow
+
+    def checkedout(self):
+        return self._pool.maxsize - self._pool.qsize() + self._overflow
+
+
+class NullPool(Pool):
+
+    """A Pool which does not pool connections.
+
+    Instead it literally opens and closes the underlying DB-API connection
+    per each connection open/close.
+
+    Reconnect-related functions such as ``recycle`` and connection
+    invalidation are not supported by this Pool implementation, since
+    no connections are held persistently.
+
+    .. versionchanged:: 0.7
+        :class:`.NullPool` is used by the SQlite dialect automatically
+        when a file-based database is used. See :ref:`sqlite_toplevel`.
+
+    """
+
+    def status(self):
+        return "NullPool"
+
+    def _do_return_conn(self, conn):
+        conn.close()
+
+    def _do_get(self):
+        return self._create_connection()
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+
+        return self.__class__(self._creator,
+                              recycle=self._recycle,
+                              echo=self.echo,
+                              logging_name=self._orig_logging_name,
+                              use_threadlocal=self._use_threadlocal,
+                              reset_on_return=self._reset_on_return,
+                              _dispatch=self.dispatch,
+                              dialect=self._dialect)
+
+    def dispose(self):
+        pass
+
+
+class SingletonThreadPool(Pool):
+
+    """A Pool that maintains one connection per thread.
+
+    Maintains one connection per each thread, never moving a connection to a
+    thread other than the one which it was created in.
+
+    .. warning::  the :class:`.SingletonThreadPool` will call ``.close()``
+       on arbitrary connections that exist beyond the size setting of
+       ``pool_size``, e.g. if more unique **thread identities**
+       than what ``pool_size`` states are used.   This cleanup is
+       non-deterministic and not sensitive to whether or not the connections
+       linked to those thread identities are currently in use.
+
+       :class:`.SingletonThreadPool` may be improved in a future release,
+       however in its current status it is generally used only for test
+       scenarios using a SQLite ``:memory:`` database and is not recommended
+       for production use.
+
+
+    Options are the same as those of :class:`.Pool`, as well as:
+
+    :param pool_size: The number of threads in which to maintain connections
+        at once.  Defaults to five.
+
+    :class:`.SingletonThreadPool` is used by the SQLite dialect
+    automatically when a memory-based database is used.
+    See :ref:`sqlite_toplevel`.
+
+    """
+
+    def __init__(self, creator, pool_size=5, **kw):
+        kw['use_threadlocal'] = True
+        Pool.__init__(self, creator, **kw)
+        self._conn = threading.local()
+        self._all_conns = set()
+        self.size = pool_size
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+        return self.__class__(self._creator,
+                              pool_size=self.size,
+                              recycle=self._recycle,
+                              echo=self.echo,
+                              logging_name=self._orig_logging_name,
+                              use_threadlocal=self._use_threadlocal,
+                              reset_on_return=self._reset_on_return,
+                              _dispatch=self.dispatch,
+                              dialect=self._dialect)
+
+    def dispose(self):
+        """Dispose of this pool."""
+
+        for conn in self._all_conns:
+            try:
+                conn.close()
+            except Exception:
+                # pysqlite won't even let you close a conn from a thread
+                # that didn't create it
+                pass
+
+        self._all_conns.clear()
+
+    def _cleanup(self):
+        while len(self._all_conns) >= self.size:
+            c = self._all_conns.pop()
+            c.close()
+
+    def status(self):
+        return "SingletonThreadPool id:%d size: %d" % \
+            (id(self), len(self._all_conns))
+
+    def _do_return_conn(self, conn):
+        pass
+
+    def _do_get(self):
+        try:
+            c = self._conn.current()
+            if c:
+                return c
+        except AttributeError:
+            pass
+        c = self._create_connection()
+        self._conn.current = weakref.ref(c)
+        if len(self._all_conns) >= self.size:
+            self._cleanup()
+        self._all_conns.add(c)
+        return c
+
+
+class StaticPool(Pool):
+
+    """A Pool of exactly one connection, used for all requests.
+
+    Reconnect-related functions such as ``recycle`` and connection
+    invalidation (which is also used to support auto-reconnect) are not
+    currently supported by this Pool implementation but may be implemented
+    in a future release.
+
+    """
+
+    @util.memoized_property
+    def _conn(self):
+        return self._creator()
+
+    @util.memoized_property
+    def connection(self):
+        return _ConnectionRecord(self)
+
+    def status(self):
+        return "StaticPool"
+
+    def dispose(self):
+        if '_conn' in self.__dict__:
+            self._conn.close()
+            self._conn = None
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+        return self.__class__(creator=self._creator,
+                              recycle=self._recycle,
+                              use_threadlocal=self._use_threadlocal,
+                              reset_on_return=self._reset_on_return,
+                              echo=self.echo,
+                              logging_name=self._orig_logging_name,
+                              _dispatch=self.dispatch,
+                              dialect=self._dialect)
+
+    def _create_connection(self):
+        return self._conn
+
+    def _do_return_conn(self, conn):
+        pass
+
+    def _do_get(self):
+        return self.connection
+
+
+class AssertionPool(Pool):
+
+    """A :class:`.Pool` that allows at most one checked out connection at
+    any given time.
+
+    This will raise an exception if more than one connection is checked out
+    at a time.  Useful for debugging code that is using more connections
+    than desired.
+
+    .. versionchanged:: 0.7
+        :class:`.AssertionPool` also logs a traceback of where
+        the original connection was checked out, and reports
+        this in the assertion error raised.
+
+    """
+
+    def __init__(self, *args, **kw):
+        self._conn = None
+        self._checked_out = False
+        self._store_traceback = kw.pop('store_traceback', True)
+        self._checkout_traceback = None
+        Pool.__init__(self, *args, **kw)
+
+    def status(self):
+        return "AssertionPool"
+
+    def _do_return_conn(self, conn):
+        if not self._checked_out:
+            raise AssertionError("connection is not checked out")
+        self._checked_out = False
+        assert conn is self._conn
+
+    def dispose(self):
+        self._checked_out = False
+        if self._conn:
+            self._conn.close()
+
+    def recreate(self):
+        self.logger.info("Pool recreating")
+        return self.__class__(self._creator, echo=self.echo,
+                              logging_name=self._orig_logging_name,
+                              _dispatch=self.dispatch,
+                              dialect=self._dialect)
+
+    def _do_get(self):
+        if self._checked_out:
+            if self._checkout_traceback:
+                suffix = ' at:\n%s' % ''.join(
+                    chop_traceback(self._checkout_traceback))
+            else:
+                suffix = ''
+            raise AssertionError("connection is already checked out" + suffix)
+
+        if not self._conn:
+            self._conn = self._create_connection()
+
+        self._checked_out = True
+        if self._store_traceback:
+            self._checkout_traceback = traceback.format_stack()
+        return self._conn