Add server address and version

README.rst

@@ -33,7 +33,6 @@ Example Usage
         with session.begin_transaction() as write_tx:
             write_tx.run("CREATE (a:Person {name:{name},age:{age}})", name="Alice", age=33)
             write_tx.run("CREATE (a:Person {name:{name},age:{age}})", name="Bob", age=44)
-            write_tx.success = True
 
         with session.begin_transaction() as read_tx:
             result = read_tx.run("MATCH (a:Person) RETURN a.name AS name, a.age AS age")

neo4j/v1/bolt.py

@@ -29,7 +29,7 @@
 from __future__ import division
 
 from base64 import b64encode
-from collections import deque
+from collections import deque, namedtuple
 from io import BytesIO
 import logging
 from os import makedirs, open as os_open, write as os_write, close as os_close, O_CREAT, O_APPEND, O_WRONLY
@@ -81,12 +81,16 @@
 log_error = log.error
 
 
+Address = namedtuple("Address", ["host", "port"])
+ServerInfo = namedtuple("ServerInfo", ["address", "version"])
+
+
 class BufferingSocket(object):
 
     def __init__(self, connection):
         self.connection = connection
         self.socket = connection.socket
-        self.address = self.socket.getpeername()
+        self.address = Address(*self.socket.getpeername())
         self.buffer = bytearray()
 
     def fill(self):
@@ -132,7 +136,7 @@ class ChunkChannel(object):
 
     def __init__(self, sock):
         self.socket = sock
-        self.address = sock.getpeername()
+        self.address = Address(*sock.getpeername())
         self.raw = BytesIO()
         self.output_buffer = []
         self.output_size = 0
@@ -206,6 +210,22 @@ def on_ignored(self, metadata=None):
         pass
 
 
+class InitResponse(Response):
+
+    def on_success(self, metadata):
+        super(InitResponse, self).on_success(metadata)
+        connection = self.connection
+        address = Address(*connection.socket.getpeername())
+        version = metadata.get("server")
+        connection.server = ServerInfo(address, version)
+
+    def on_failure(self, metadata):
+        code = metadata.get("code")
+        error = (Unauthorized if code == "Neo.ClientError.Security.Unauthorized" else
+                 ServiceUnavailable)
+        raise error(metadata.get("message", "INIT failed"))
+
+
 class Connection(object):
     """ Server connection for Bolt protocol v1.
 
@@ -216,20 +236,25 @@ class Connection(object):
     .. note:: logs at INFO level
     """
 
+    in_use = False
+
+    closed = False
+
+    defunct = False
+
     #: The pool of which this connection is a member
     pool = None
 
+    #: Server version details
+    server = None
+
     def __init__(self, sock, **config):
         self.socket = sock
         self.buffering_socket = BufferingSocket(self)
-        self.address = sock.getpeername()
         self.channel = ChunkChannel(sock)
         self.packer = Packer(self.channel)
         self.unpacker = Unpacker()
         self.responses = deque()
-        self.in_use = False
-        self.closed = False
-        self.defunct = False
 
         # Determine the user agent and ensure it is a Unicode value
         user_agent = config.get("user_agent", DEFAULT_USER_AGENT)
@@ -246,19 +271,9 @@ def __init__(self, sock, **config):
         # Pick up the server certificate, if any
         self.der_encoded_server_certificate = config.get("der_encoded_server_certificate")
 
-        def on_failure(metadata):
-            code = metadata.get("code")
-            error = (Unauthorized if code == "Neo.ClientError.Security.Unauthorized" else
-                     ServiceUnavailable)
-            raise error(metadata.get("message", "INIT failed"))
-
-        response = Response(self)
-        response.on_failure = on_failure
-
+        response = InitResponse(self)
         self.append(INIT, (self.user_agent, self.auth_dict), response=response)
-        self.send()
-        while not response.complete:
-            self.fetch()
+        self.sync()
 
     def __del__(self):
         self.close()
@@ -316,18 +331,18 @@ def send(self):
         """ Send all queued messages to the server.
         """
         if self.closed:
-            raise ServiceUnavailable("Failed to write to closed connection %r" % (self.address,))
+            raise ServiceUnavailable("Failed to write to closed connection %r" % (self.server.address,))
         if self.defunct:
-            raise ServiceUnavailable("Failed to write to defunct connection %r" % (self.address,))
+            raise ServiceUnavailable("Failed to write to defunct connection %r" % (self.server.address,))
         self.channel.send()
 
     def fetch(self):
         """ Receive exactly one message from the server.
         """
         if self.closed:
-            raise ServiceUnavailable("Failed to read from closed connection %r" % (self.address,))
+            raise ServiceUnavailable("Failed to read from closed connection %r" % (self.server.address,))
         if self.defunct:
-            raise ServiceUnavailable("Failed to read from defunct connection %r" % (self.address,))
+            raise ServiceUnavailable("Failed to read from defunct connection %r" % (self.server.address,))
         try:
             message_data = self.buffering_socket.read_message()
         except ProtocolError:
@@ -367,7 +382,10 @@ def fetch(self):
         else:
             raise ProtocolError("Unexpected response message with signature %02X" % signature)
 
-    def fetch_all(self):
+    def sync(self):
+        """ Send and fetch all outstanding messages.
+        """
+        self.send()
         while self.responses:
             response = self.responses[0]
             while not response.complete:

neo4j/v1/routing.py

@@ -22,7 +22,7 @@
 from threading import Lock
 from time import clock
 
-from .bolt import ConnectionPool
+from .bolt import Address, ConnectionPool
 from .compat.collections import MutableSet, OrderedDict
 from .exceptions import CypherError, ProtocolError, ServiceUnavailable
 
@@ -94,7 +94,7 @@ def parse_address(cls, address):
         """ Convert an address string to a tuple.
         """
         host, _, port = address.partition(":")
-        return host, int(port)
+        return Address(host, int(port))
 
     @classmethod
     def parse_routing_info(cls, records):

neo4j/v1/session.py

@@ -241,6 +241,8 @@ class Session(object):
 
     transaction = None
 
+    last_bookmark = None
+
     def __init__(self, connection, access_mode=None):
         self.connection = connection
         self.access_mode = access_mode
@@ -265,6 +267,8 @@ def run(self, statement, parameters=None, **kwparameters):
         :return: Cypher result
         :rtype: :class:`.StatementResult`
         """
+        self.last_bookmark = None
+
         statement = _norm_statement(statement)
         parameters = _norm_parameters(parameters, **kwparameters)
 
@@ -301,13 +305,15 @@ def close(self):
             self.transaction.close()
         if self.connection:
             if not self.connection.closed:
-                self.connection.fetch_all()
+                self.connection.sync()
             self.connection.in_use = False
             self.connection = None
 
-    def begin_transaction(self):
+    def begin_transaction(self, bookmark=None):
         """ Create a new :class:`.Transaction` within this session.
 
+        :param bookmark: a bookmark to which the server should
+                         synchronise before beginning the transaction
         :return: new :class:`.Transaction` instance.
         """
         if self.transaction:
@@ -316,15 +322,23 @@ def begin_transaction(self):
         def clear_transaction():
             self.transaction = None
 
-        self.run("BEGIN")
+        parameters = {}
+        if bookmark is not None:
+            parameters["bookmark"] = bookmark
+
+        self.run("BEGIN", parameters)
         self.transaction = Transaction(self, on_close=clear_transaction)
         return self.transaction
 
     def commit_transaction(self):
-        self.run("COMMIT")
+        result = self.run("COMMIT")
+        self.connection.sync()
+        summary = result.summary()
+        self.last_bookmark = summary.metadata.get("bookmark")
 
     def rollback_transaction(self):
         self.run("ROLLBACK")
+        self.connection.sync()
 
 
 class Transaction(object):
@@ -342,7 +356,7 @@ class Transaction(object):
     #: and rolled back otherwise. This attribute can be set in user code
     #: multiple times before a transaction completes with only the final
     #: value taking effect.
-    success = False
+    success = None
 
     #: Indicator to show whether the transaction has been closed, either
     #: with commit or rollback.
@@ -356,8 +370,8 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
-        if exc_value:
-            self.success = False
+        if self.success is None:
+            self.success = not bool(exc_type)
         self.close()
 
     def run(self, statement, parameters=None, **kwparameters):

test/test_driver.py

@@ -159,7 +159,7 @@ def test_should_be_able_to_read(self):
                     result = session.run("RETURN $x", {"x": 1})
                     for record in result:
                         assert record["x"] == 1
-                    assert session.connection.address == ('127.0.0.1', 9004)
+                    assert session.connection.server.address == ('127.0.0.1', 9004)
 
     def test_should_be_able_to_write(self):
         with StubCluster({9001: "router.script", 9006: "create_a.script"}):
@@ -168,7 +168,7 @@ def test_should_be_able_to_write(self):
                 with driver.session(WRITE_ACCESS) as session:
                     result = session.run("CREATE (a $x)", {"x": {"name": "Alice"}})
                     assert not list(result)
-                    assert session.connection.address == ('127.0.0.1', 9006)
+                    assert session.connection.server.address == ('127.0.0.1', 9006)
 
     def test_should_be_able_to_write_as_default(self):
         with StubCluster({9001: "router.script", 9006: "create_a.script"}):
@@ -177,7 +177,7 @@ def test_should_be_able_to_write_as_default(self):
                 with driver.session() as session:
                     result = session.run("CREATE (a $x)", {"x": {"name": "Alice"}})
                     assert not list(result)
-                    assert session.connection.address == ('127.0.0.1', 9006)
+                    assert session.connection.server.address == ('127.0.0.1', 9006)
 
     def test_routing_disconnect_on_run(self):
         with StubCluster({9001: "router.script", 9004: "disconnect_on_run.script"}):

test/test_routing.py

@@ -575,7 +575,7 @@ def test_connected_to_reader(self):
             with RoutingConnectionPool(connector, address) as pool:
                 assert not pool.routing_table.is_fresh()
                 connection = pool.acquire_for_read()
-                assert connection.address in pool.routing_table.readers
+                assert connection.server.address in pool.routing_table.readers
 
     def test_should_retry_if_first_reader_fails(self):
         with StubCluster({9001: "router.script",
@@ -605,7 +605,7 @@ def test_connected_to_writer(self):
             with RoutingConnectionPool(connector, address) as pool:
                 assert not pool.routing_table.is_fresh()
                 connection = pool.acquire_for_write()
-                assert connection.address in pool.routing_table.writers
+                assert connection.server.address in pool.routing_table.writers
 
     def test_should_retry_if_first_writer_fails(self):
         with StubCluster({9001: "router_with_multiple_writers.script",