diff --git a/README.rst b/README.rst index c894fb85d..8d88676ba 100644 --- a/README.rst +++ b/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") diff --git a/neo4j/v1/bolt.py b/neo4j/v1/bolt.py index f237ed564..f85e9e19d 100644 --- a/neo4j/v1/bolt.py +++ b/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: diff --git a/neo4j/v1/routing.py b/neo4j/v1/routing.py index a408fdfe1..a58c758b2 100644 --- a/neo4j/v1/routing.py +++ b/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): diff --git a/neo4j/v1/session.py b/neo4j/v1/session.py index 2b96c4352..b92fc2dcd 100644 --- a/neo4j/v1/session.py +++ b/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): diff --git a/test/test_driver.py b/test/test_driver.py index 55580f6fb..76de32c48 100644 --- a/test/test_driver.py +++ b/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"}): diff --git a/test/test_routing.py b/test/test_routing.py index 8a30aa3e8..b4c60b12f 100644 --- a/test/test_routing.py +++ b/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", diff --git a/test/test_session.py b/test/test_session.py index b8d2022c7..e3b9024ce 100644 --- a/test/test_session.py +++ b/test/test_session.py @@ -19,7 +19,10 @@ # limitations under the License. from mock import patch +from unittest import skipUnless +from uuid import uuid4 +from neo4j.v1 import READ_ACCESS, WRITE_ACCESS from neo4j.v1.exceptions import CypherError, ResultError from neo4j.v1.session import GraphDatabase, basic_auth, Record from neo4j.v1.types import Node, Relationship, Path @@ -31,6 +34,21 @@ AUTH_TOKEN = basic_auth("neotest", "neotest") +def get_server_version(): + driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) + with driver.session() as session: + full_version = session.connection.server.version + if full_version is None: + return "Neo4j", (3, 0), () + product, _, tagged_version = full_version.partition("/") + tags = tagged_version.split("-") + version = map(int, tags[0].split(".")) + return product, tuple(version), tuple(tags[1:]) + + +SERVER_PRODUCT, SERVER_VERSION, SERVER_TAGS = get_server_version() + + class AutoCommitTransactionTestCase(ServerTestCase): def setUp(self): @@ -415,17 +433,60 @@ def test_can_rollback_transaction_using_with_block(self): tx.run("MATCH (a) WHERE id(a) = {n} " "SET a.foo = {foo}", {"n": node_id, "foo": "bar"}) + tx.success = False + # Check the property value result = session.run("MATCH (a) WHERE id(a) = {n} " "RETURN a.foo", {"n": node_id}) assert len(list(result)) == 0 +class BookmarkingTestCase(ServerTestCase): + + def setUp(self): + self.driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) + + def tearDown(self): + self.driver.close() + + @skipUnless(SERVER_VERSION >= (3, 1), "Bookmarking is not supported by this version of Neo4j") + def test_can_obtain_bookmark_after_commit(self): + with self.driver.session() as session: + with session.begin_transaction() as tx: + tx.run("RETURN 1") + assert session.last_bookmark is not None + + @skipUnless(SERVER_VERSION >= (3, 1), "Bookmarking is not supported by this version of Neo4j") + def test_can_pass_bookmark_into_next_transaction(self): + unique_id = uuid4().hex + + with self.driver.session(WRITE_ACCESS) as session: + with session.begin_transaction() as tx: + tx.run("CREATE (a:Thing {uuid:$uuid})", uuid=unique_id) + bookmark = session.last_bookmark + + assert bookmark is not None + + with self.driver.session(READ_ACCESS) as session: + with session.begin_transaction(bookmark) as tx: + result = tx.run("MATCH (a:Thing {uuid:$uuid}) RETURN a", uuid=unique_id) + record_list = list(result) + assert len(record_list) == 1 + record = record_list[0] + assert len(record) == 1 + thing = record[0] + assert isinstance(thing, Node) + assert thing["uuid"] == unique_id + + class ResultConsumptionTestCase(ServerTestCase): def setUp(self): self.driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN, encrypted=False) + def tearDown(self): + self.driver.close() + def test_can_consume_result_immediately(self): session = self.driver.session() tx = session.begin_transaction() @@ -564,3 +625,39 @@ def test_peek_at_different_stages(self): # ...when none should follow with self.assertRaises(ResultError): result.peek() + + +class SessionCommitTestCase(ServerTestCase): + + def setUp(self): + self.driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN) + + def tearDown(self): + self.driver.close() + + def test_should_sync_after_commit(self): + with self.driver.session() as session: + tx = session.begin_transaction() + result = tx.run("RETURN 1") + tx.commit() + buffer = result._buffer + assert len(buffer) == 1 + assert buffer[0][0] == 1 + + +class SessionRollbackTestCase(ServerTestCase): + + def setUp(self): + self.driver = GraphDatabase.driver(BOLT_URI, auth=AUTH_TOKEN) + + def tearDown(self): + self.driver.close() + + def test_should_sync_after_rollback(self): + with self.driver.session() as session: + tx = session.begin_transaction() + result = tx.run("RETURN 1") + tx.rollback() + buffer = result._buffer + assert len(buffer) == 1 + assert buffer[0][0] == 1