README.rst

@@ -21,7 +21,7 @@ Key features include:
 - Ability to directly access psycopg2 ``connection`` and ``cursor`` objects
 - Internal connection pooling
-|Version| |Downloads| |Status| |Coverage| |CodeClimate| |License| |PythonVersions|
+|Version| |Status| |Coverage| |CodeClimate| |License| |PythonVersions|
 Documentation
 -------------
@@ -179,9 +179,6 @@ main GitHub repository of Queries as tags prior to version 1.2.0.
 .. |Coverage| image:: https://img.shields.io/codecov/c/github/gmr/queries.svg?
    :target: https://codecov.io/github/gmr/queries?branch=master
-.. |Downloads| image:: https://img.shields.io/pypi/dm/queries.svg?
-   :target: https://pypi.python.org/pypi/queries
-
 .. |License| image:: https://img.shields.io/github/license/gmr/queries.svg?
    :target: https://github.com/gmr/queries

docs/history.rst

 Version History
 ===============
-- Next Release
-  - Log a warning when a tornado_session.Result is ``__del__'d`` without ``free`` being called.
+- 1.10.3 2017-11-01
+  - Remove the functionality from ``TornadoSession.validate`` and make it raise a ``DeprecationWarning``
+  - Catch the ``KeyError`` raised when ``PoolManager.clean()`` is invoked for a pool that doesn't exist
+- 1.10.2 2017-10-26
+  - Ensure the pool exists when executing a query in TornadoSession, the new timeout behavior prevented that from happening.
+- 1.10.1 2017-10-24
+  - Use an absolute time in the call to ``add_timeout``
+- 1.10.0 2017-09-27
+  - Free when tornado_session.Result is ``__del__'d`` without ``free`` being called.
+  - Auto-clean the pool after Results.free TTL+1 in tornado_session.TornadoSession
+  - Dont raise NotImplementedError in Results.free for synchronous use, just treat as a noop
 - 1.9.1 2016-10-25
   - Add better exception handling around connections and getting the logged in user
 - 1.9.0 2016-07-01

docs/index.rst

@@ -3,7 +3,7 @@ Queries: PostgreSQL Simplified
 *Queries* is a BSD licensed opinionated wrapper of the psycopg2_ library for
 interacting with PostgreSQL.
-|Version| |Downloads| |License| |PythonVersions|
+|Version| |License| |PythonVersions|
 The popular psycopg2_ package is a full-featured python client. Unfortunately
 as a developer, you're often repeating the same steps to get started with your

queries/__init__.py

@@ -8,7 +8,7 @@ The core `queries.Queries` class will automatically register support for UUIDs,
 Unicode and Unicode arrays.
 """
-__version__ = '1.9.1'
+__version__ = '1.10.3'
 version = __version__
 import logging

queries/pool.py

@@ -132,11 +132,13 @@ class Pool(object):
     def __init__(self,
                  pool_id,
                  idle_ttl=DEFAULT_IDLE_TTL,
-                 max_size=DEFAULT_MAX_SIZE):
+                 max_size=DEFAULT_MAX_SIZE,
+                 time_method=None):
         self.connections = {}
         self._id = pool_id
         self.idle_ttl = idle_ttl
         self.max_size = max_size
+        self.time_method = time_method or time.time
     def __contains__(self, connection):
         """Return True if the pool contains the connection"""
@@ -207,7 +209,7 @@ class Pool(object):
         if self.idle_connections == list(self.connections.values()):
             with self._lock:
-                self.idle_start = time.time()
+                self.idle_start = self.time_method()
         LOGGER.debug('Pool %s freed connection %s', self.id, id(connection))
     def get(self, session):
@@ -258,7 +260,7 @@ class Pool(object):
         """
         if self.idle_start is None:
             return 0
-        return time.time() - self.idle_start
+        return self.time_method() - self.idle_start
     @property
     def is_full(self):
@@ -405,15 +407,17 @@ class PoolManager(object):
         """
         with cls._lock:
-            cls._ensure_pool_exists(pid)
+            try:
+                cls._ensure_pool_exists(pid)
+            except KeyError:
+                LOGGER.debug('Pool clean invoked against missing pool %s', pid)
+                return
             cls._pools[pid].clean()
-
-            # If the pool has no open connections, remove it
-            if not len(cls._pools[pid]):
-                del cls._pools[pid]
+            cls._maybe_remove_pool(pid)
     @classmethod
-    def create(cls, pid, idle_ttl=DEFAULT_IDLE_TTL, max_size=DEFAULT_MAX_SIZE):
+    def create(cls, pid, idle_ttl=DEFAULT_IDLE_TTL, max_size=DEFAULT_MAX_SIZE,
+               time_method=None):
         """Create a new pool, with the ability to pass in values to override
         the default idle TTL and the default maximum size.
@@ -426,6 +430,8 @@ class PoolManager(object):
         :param str pid: The pool ID
         :param int idle_ttl: Time in seconds for the idle TTL
         :param int max_size: The maximum pool size
+        :param callable time_method: Override the use of :py:meth:`time.time`
+            method for time values.
         :raises: KeyError
         """
@@ -433,7 +439,7 @@ class PoolManager(object):
             raise KeyError('Pool %s already exists' % pid)
         with cls._lock:
             LOGGER.debug("Creating Pool: %s (%i/%i)", pid, idle_ttl, max_size)
-            cls._pools[pid] = Pool(pid, idle_ttl, max_size)
+            cls._pools[pid] = Pool(pid, idle_ttl, max_size, time_method)
     @classmethod
     def get(cls, pid, session):
@@ -595,6 +601,16 @@ class PoolManager(object):
         if pid not in cls._pools:
             raise KeyError('Pool %s has not been created' % pid)
+    @classmethod
+    def _maybe_remove_pool(cls, pid):
+        """If the pool has no open connections, remove it
+
+        :param str pid: The pool id to clean
+
+        """
+        if not len(cls._pools[pid]):
+            del cls._pools[pid]
+
 class QueriesException(Exception):
     """Base Exception for all other Queries exceptions"""

queries/results.py

@@ -92,7 +92,7 @@ class Results(object):
         connections.
         """
-        raise NotImplementedError
+        LOGGER.debug('Invoking synchronous free has no effect')
     def items(self):
         """Return all of the rows that are in the result set.

queries/session.py

@@ -30,15 +30,10 @@ import psycopg2
 from psycopg2 import extensions
 from psycopg2 import extras
-from queries import pool
-from queries import results
-from queries import utils
+from queries import pool, results, utils, DEFAULT_URI, PYPY
 LOGGER = logging.getLogger(__name__)
-from queries import DEFAULT_URI
-from queries import PYPY
-
 DEFAULT_ENCODING = 'UTF8'

queries/tornado_session.py

@@ -25,19 +25,14 @@ Example Use:
 import logging
 import socket
-from psycopg2 import extensions
-from psycopg2 import extras
-
-from tornado import concurrent
-from tornado import gen
-from tornado import ioloop
 import psycopg2
+from psycopg2 import extras, extensions
+from tornado import concurrent, ioloop
 from queries import pool
 from queries import results
 from queries import session
 from queries import utils
-
 from queries import DEFAULT_URI
 from queries import PYPY
@@ -109,7 +104,6 @@ class Results(results.Results):
         self._fd = fd
         self._freed = False
-    @gen.coroutine
     def free(self):
         """Release the results and connection lock from the TornadoSession
         object. This **must** be called after you finish processing the results
@@ -124,7 +118,8 @@ class Results(results.Results):
     def __del__(self):
         if not self._freed:
-            LOGGER.warning('%s not freed - %r', self.__class__.__name__, self)
+            LOGGER.warning('Auto-freeing result on deletion')
+            self.free()
 class TornadoSession(session.Session):
@@ -161,16 +156,21 @@ class TornadoSession(session.Session):
         """
         self._connections = dict()
-        self._futures = dict()
-
+        self._cleanup_callback = None
         self._cursor_factory = cursor_factory
-        self._ioloop = io_loop or ioloop.IOLoop.instance()
+        self._futures = dict()
+        self._ioloop = io_loop or ioloop.IOLoop.current()
         self._pool_manager = pool.PoolManager.instance()
+        self._pool_max_size = pool_max_size
+        self._pool_idle_ttl = pool_idle_ttl
         self._uri = uri
-        # Ensure the pool exists in the pool manager
+        self._ensure_pool_exists()
+    def _ensure_pool_exists(self):
+        """Create the pool in the pool manager if it does not exist."""
         if self.pid not in self._pool_manager:
-            self._pool_manager.create(self.pid, pool_idle_ttl, pool_max_size)
+            self._pool_manager.create(self.pid, self._pool_idle_ttl,
+                                      self._pool_max_size, self._ioloop.time)
     @property
     def connection(self):
@@ -234,38 +234,17 @@ class TornadoSession(session.Session):
     def validate(self):
         """Validate the session can connect or has open connections to
-        PostgreSQL
+        PostgreSQL. As of ``1.10.3``
+
+        .. deprecated:: 1.10.3
+           As of 1.10.3, this method only raises a
+           :py:exception:`DeprecationWarning`.
         :rtype: bool
+        :raises: DeprecationWarning
         """
-        future = concurrent.TracebackFuture()
-
-        def on_connected(cf):
-            if cf.exception():
-                future.set_exception(cf.exception())
-                return
-
-            connection = cf.result()
-            fd = connection.fileno()
-
-            # The connection would have been added to the pool manager, free it
-            self._pool_manager.free(self.pid, connection)
-            self._ioloop.remove_handler(fd)
-
-            if fd in self._connections:
-                del self._connections[fd]
-            if fd in self._futures:
-                del self._futures[fd]
-
-            # Return the success in validating the connection
-            future.set_result(True)
-
-        # Grab a connection to PostgreSQL
-        self._ioloop.add_future(self._connect(), on_connected)
-
-        # Return the future for the query result
-        return future
+        raise DeprecationWarning('All functionality removed from this method')
     def _connect(self):
         """Connect to PostgreSQL, either by reusing a connection from the pool
@@ -415,6 +394,9 @@ class TornadoSession(session.Session):
             func = getattr(cursor, method)
             func(query, parameters)
+        # Ensure the pool exists for the connection
+        self._ensure_pool_exists()
+
         # Grab a connection to PostgreSQL
         self._ioloop.add_future(self._connect(), on_connected)
@@ -438,16 +420,25 @@ class TornadoSession(session.Session):
         self._pool_manager.free(self.pid, self._connections[fd])
         self._ioloop.remove_handler(fd)
+        # If the cleanup callback exists, remove it
+        if self._cleanup_callback:
+            self._ioloop.remove_timeout(self._cleanup_callback)
+
+        # Create a new cleanup callback to clean the pool of idle connections
+        self._cleanup_callback = self._ioloop.add_timeout(
+            self._ioloop.time() + self._pool_idle_ttl + 1,
+            self._pool_manager.clean, self.pid)
+
         if fd in self._connections:
             del self._connections[fd]
         if fd in self._futures:
             del self._futures[fd]
-    def _on_io_events(self, fd=None, events=None):
+    def _on_io_events(self, fd=None, _events=None):
         """Invoked by Tornado's IOLoop when there are events for the fd
         :param int fd: The file descriptor for the event
-        :param int events: The events raised
+        :param int _events: The events raised
         """
         if fd not in self._connections:

setup.cfg

+[flake8]
+exclude = .git,build,dist,docs,env
+
 [nosetests]
-with-coverage=1
-cover-package=queries
-cover-branches=1
-cover-erase=1
+cover-branches = 1
+cover-erase = 1
+cover-html = 1
+cover-html-dir =  build/coverage
+cover-package = queries
+logging-level = DEBUG
+verbosity = 2
+with-coverage = 1

setup.py

@@ -20,16 +20,15 @@ classifiers = ['Development Status :: 5 - Production/Stable',
                'Programming Language :: Python :: 2.6',
                'Programming Language :: Python :: 2.7',
                'Programming Language :: Python :: 3',
-               'Programming Language :: Python :: 3.3',
-               'Programming Language :: Python :: 3.4',
                'Programming Language :: Python :: 3.5',
+               'Programming Language :: Python :: 3.6',
                'Programming Language :: Python :: Implementation :: CPython',
                'Programming Language :: Python :: Implementation :: PyPy',
                'Topic :: Database',
                'Topic :: Software Development :: Libraries']
 setup(name='queries',
-      version='1.9.1',
+      version='1.10.3',
       description="Simplified PostgreSQL client built upon Psycopg2",
       maintainer="Gavin M. Roy",
       maintainer_email="gavinmroy@gmail.com",

test-requirements.txt

 coverage
+flake8
 mock
 nose
 codecov

tests/pool_manager_tests.py

@@ -51,9 +51,13 @@ class ManagerTests(unittest.TestCase):
         psycopg2_conn = mock.Mock()
         self.assertRaises(KeyError, self.manager.add, pid, psycopg2_conn)
-    def test_clean_ensures_pool_exists(self):
+    def test_ensures_pool_exists_raises_key_error(self):
         pid = str(uuid.uuid4())
-        self.assertRaises(KeyError, self.manager.clean, pid)
+        self.assertRaises(KeyError, self.manager._ensure_pool_exists, pid)
+
+    def test_clean_ensures_pool_exists_catches_key_error(self):
+        pid = str(uuid.uuid4())
+        self.assertIsNone(self.manager.clean(pid))
     def test_clean_invokes_pool_clean(self):
         pid = str(uuid.uuid4())

tests/results_tests.py

@@ -45,7 +45,8 @@ class ResultsTestCase(unittest.TestCase):
         self.cursor.rowcount = 0
         with mock.patch.object(self.obj, '_rewind') as rewind:
             [x for x in self.obj]
-            assert not rewind.called, '_rewind should not be called on empty result'
+            assert not rewind.called, \
+                '_rewind should not be called on empty result'
     def test_iter_rewinds(self):
         self.cursor.__iter__ = mock.Mock(return_value=iter([1, 2, 3]))
@@ -102,15 +103,13 @@ class ResultsTestCase(unittest.TestCase):
         self.cursor.rowcount = 2
         self.assertEqual(self.obj.count(), 2)
-    def test_free_raises_exception(self):
-        self.assertRaises(NotImplementedError, self.obj.free)
-
     def test_items_returns_on_empty(self):
         self.cursor.rowcount = 0
         self.cursor.scroll = mock.Mock()
         self.cursor.fetchall = mock.Mock()
         self.obj.items()
-        assert not self.cursor.scroll.called, 'Cursor.scroll should not be called on empty result'
+        assert not self.cursor.scroll.called, \
+            'Cursor.scroll should not be called on empty result'
     def test_items_invokes_scroll(self):
         self.cursor.scroll = mock.Mock()

tests/tornado_session_tests.py

@@ -47,11 +47,11 @@ class SessionInitTests(unittest.TestCase):
     def setUp(self):
         self.obj = tornado_session.TornadoSession()
-    #def test_creates_empty_callback_dict(self):
-    #    self.assertDictEqual(self.obj._futures, {})
-   # def test_creates_empty_connections_dict(self):
-   #     self.assertDictEqual(self.obj._connections, {})
+    def test_creates_empty_callback_dict(self):
+        self.assertDictEqual(self.obj._futures, {})
+    def test_creates_empty_connections_dict(self):
+        self.assertDictEqual(self.obj._connections, {})
     def test_sets_default_cursor_factory(self):
         self.assertEqual(self.obj._cursor_factory, extras.RealDictCursor)
@@ -105,8 +105,8 @@ class SessionConnectTests(testing.AsyncTestCase):
         conn = yield self.obj._connect()
         self.obj._pool_manager.add(self.obj.pid, conn)
         with mock.patch.object(self.obj._pool_manager, 'get') as get:
-            with mock.patch.object(self.io_loop, 'add_handler') as add_handler:
-                second_result = yield self.obj._connect()
+            with mock.patch.object(self.io_loop, 'add_handler'):
+                yield self.obj._connect()
                 get.assert_called_once_with(self.obj.pid, self.obj)
     @testing.gen_test
@@ -116,7 +116,7 @@ class SessionConnectTests(testing.AsyncTestCase):
         with mock.patch.object(self.obj._pool_manager, 'get') as get:
             get.return_value = self.conn
             with mock.patch.object(self.io_loop, 'add_handler') as add_handler:
-                second_result = yield self.obj._connect()
+                yield self.obj._connect()
                 add_handler.assert_called_once_with(self.conn.fileno(),
                                                     self.obj._on_io_events,
                                                     ioloop.IOLoop.WRITE)
@@ -232,5 +232,5 @@ class SessionPublicMethodTests(testing.AsyncTestCase):
                 future.set_result(connection)
                 _connect.return_value = future
                 obj = tornado_session.TornadoSession(io_loop=self.io_loop)
-                result = yield obj.validate()
+                yield obj.validate()
                 _connect.assert_called_once_with()