Docstrings

Author: technige

neo4j/v1/bolt.py

@@ -361,24 +361,23 @@ def close(self):
 
 
 class ConnectionPool(object):
+    """ A collection of connections to one or more server addresses.
+    """
 
     def __init__(self, connector):
         self.connector = connector
-        self._connections = {}
-        self._lock = Lock()
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.close()
+        self.connections = {}
+        self.lock = Lock()
 
     def acquire(self, address):
-        with self._lock:
+        """ Acquire a connection to a given address from the pool.
+        This method is thread safe.
+        """
+        with self.lock:
             try:
-                connections = self._connections[address]
+                connections = self.connections[address]
             except KeyError:
-                connections = self._connections[address] = deque()
+                connections = self.connections[address] = deque()
             for connection in list(connections):
                 if connection.closed or connection.defunct:
                     connections.remove(connection)
@@ -392,18 +391,24 @@ def acquire(self, address):
             return connection
 
     def release(self, connection):
-        with self._lock:
+        """ Release a connection back into the pool.
+        This method is thread safe.
+        """
+        with self.lock:
             connection.in_use = False
 
     def close(self):
-        with self._lock:
-            for _, connections in self._connections.items():
+        """ Close all connections and empty the pool.
+        This method is thread safe.
+        """
+        with self.lock:
+            for _, connections in self.connections.items():
                 for connection in connections:
                     try:
                         connection.close()
                     except IOError:
                         pass
-            self._connections.clear()
+            self.connections.clear()
 
 
 class CertificateStore(object):

neo4j/v1/session.py

@@ -173,6 +173,11 @@ def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
     def session(self, access_mode=None):
+        """ Create a new session using a connection from the driver connection
+        pool. Session creation is a lightweight operation and sessions are
+        not thread safe, therefore a session should generally be short-lived
+        within a single thread.
+        """
         raise NotImplementedError()
 
     def close(self):
@@ -191,9 +196,6 @@ def __init__(self, address, **config):
         Driver.__init__(self, lambda a: connect(a, security_plan.ssl_context, **config))
 
     def session(self, access_mode=None):
-        """ Create a new session based on the graph database details
-        specified within this driver:
-        """
         return Session(self.pool.acquire(self.address))
 
 
@@ -205,6 +207,9 @@ def parse_address(address):
 
 
 class Router(object):
+    """ The `Router` class contains logic for discovering servers within a
+    cluster that supports routing.
+    """
 
     timer = monotonic
 
@@ -217,10 +222,15 @@ def __init__(self, pool, *routers):
         self.writers = RoundRobinSet()
 
     def stale(self):
+        """ Indicator for whether routing information is out of date or
+        incomplete.
+        """
         expired = self.expiry_time is None or self.expiry_time <= self.timer()
         return expired or len(self.routers) <= 1 or not self.readers or not self.writers
 
     def discover(self):
+        """ Perform cluster member discovery.
+        """
         with self.lock:
             if not self.routers:
                 raise ServiceUnavailable("No routers available")
@@ -249,11 +259,15 @@ def discover(self):
             raise ServiceUnavailable("Unable to establish routing information")
 
     def acquire_read_connection(self):
+        """ Acquire a connection to a read server.
+        """
         if self.stale():
             self.discover()
         return self.pool.acquire(next(self.readers))
 
     def acquire_write_connection(self):
+        """ Acquire a connection to a write server.
+        """
         if self.stale():
             self.discover()
         return self.pool.acquire(next(self.writers))

test/test_connection.py

@@ -18,6 +18,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from socket import create_connection
 
 from neo4j.v1.bolt import ConnectionPool
 
@@ -40,65 +41,60 @@ def close(self):
         self.socket.close()
 
 
-def connector(address):
-    from socket import create_connection
-    return QuickConnection(create_connection(address))
-
+class ConnectionPoolTestCase(ServerTestCase):
 
-def assert_pool_size(pool, address, expected_active, expected_inactive):
-    try:
-        connections = pool._connections[address]
-    except KeyError:
-        assert 0 == expected_active
-        assert 0 == expected_inactive
-    else:
-        assert len([c for c in connections if c.in_use]) == expected_active
-        assert len([c for c in connections if not c.in_use]) == expected_inactive
+    def setUp(self):
+        self.pool = ConnectionPool(lambda a: QuickConnection(create_connection(a)))
 
+    def tearDown(self):
+        self.pool.close()
 
-class ConnectionPoolTestCase(ServerTestCase):
+    def assert_pool_size(self, address, expected_active, expected_inactive):
+        try:
+            connections = self.pool.connections[address]
+        except KeyError:
+            assert 0 == expected_active
+            assert 0 == expected_inactive
+        else:
+            assert len([c for c in connections if c.in_use]) == expected_active
+            assert len([c for c in connections if not c.in_use]) == expected_inactive
 
     def test_can_acquire(self):
-        with ConnectionPool(connector) as pool:
-            address = ("127.0.0.1", 7687)
-            connection = pool.acquire(address)
-            assert connection.address == address
-            assert_pool_size(pool, address, 1, 0)
+        address = ("127.0.0.1", 7687)
+        connection = self.pool.acquire(address)
+        assert connection.address == address
+        self.assert_pool_size(address, 1, 0)
 
     def test_can_acquire_twice(self):
-        with ConnectionPool(connector) as pool:
-            address = ("127.0.0.1", 7687)
-            connection_1 = pool.acquire(address)
-            connection_2 = pool.acquire(address)
-            assert connection_1.address == address
-            assert connection_2.address == address
-            assert connection_1 is not connection_2
-            assert_pool_size(pool, address, 2, 0)
+        address = ("127.0.0.1", 7687)
+        connection_1 = self.pool.acquire(address)
+        connection_2 = self.pool.acquire(address)
+        assert connection_1.address == address
+        assert connection_2.address == address
+        assert connection_1 is not connection_2
+        self.assert_pool_size(address, 2, 0)
 
     def test_can_acquire_two_addresses(self):
-        with ConnectionPool(connector) as pool:
-            address_1 = ("127.0.0.1", 7687)
-            address_2 = ("127.0.0.1", 7474)
-            connection_1 = pool.acquire(address_1)
-            connection_2 = pool.acquire(address_2)
-            assert connection_1.address == address_1
-            assert connection_2.address == address_2
-            assert_pool_size(pool, address_1, 1, 0)
-            assert_pool_size(pool, address_2, 1, 0)
+        address_1 = ("127.0.0.1", 7687)
+        address_2 = ("127.0.0.1", 7474)
+        connection_1 = self.pool.acquire(address_1)
+        connection_2 = self.pool.acquire(address_2)
+        assert connection_1.address == address_1
+        assert connection_2.address == address_2
+        self.assert_pool_size(address_1, 1, 0)
+        self.assert_pool_size(address_2, 1, 0)
 
     def test_can_acquire_and_release(self):
-        with ConnectionPool(connector) as pool:
-            address = ("127.0.0.1", 7687)
-            connection = pool.acquire(address)
-            assert_pool_size(pool, address, 1, 0)
-            pool.release(connection)
-            assert_pool_size(pool, address, 0, 1)
+        address = ("127.0.0.1", 7687)
+        connection = self.pool.acquire(address)
+        self.assert_pool_size(address, 1, 0)
+        self.pool.release(connection)
+        self.assert_pool_size(address, 0, 1)
 
     def test_releasing_twice(self):
-        with ConnectionPool(connector) as pool:
-            address = ("127.0.0.1", 7687)
-            connection = pool.acquire(address)
-            pool.release(connection)
-            assert_pool_size(pool, address, 0, 1)
-            pool.release(connection)
-            assert_pool_size(pool, address, 0, 1)
+        address = ("127.0.0.1", 7687)
+        connection = self.pool.acquire(address)
+        self.pool.release(connection)
+        self.assert_pool_size(address, 0, 1)
+        self.pool.release(connection)
+        self.assert_pool_size(address, 0, 1)