Skip to content

Allow tx timeout to be 0 and send it #642

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jan 18, 2022
Merged

Allow tx timeout to be 0 and send it #642

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2960f05
Allow tx timeout to be 0 and send it
robsdedude Jan 11, 2022
bc6d413
Update tx timeout docs
robsdedude Jan 13, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions neo4j/_async/io/_bolt3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,11 +229,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port, " ".join(map(repr, fields)))
if query.upper() == u"COMMIT":
Expand Down Expand Up @@ -281,11 +283,13 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))

Expand Down
30 changes: 18 additions & 12 deletions neo4j/_async/io/_bolt4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,11 +181,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port, " ".join(map(repr, fields)))
if query.upper() == u"COMMIT":
Expand Down Expand Up @@ -232,11 +234,13 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))

Expand Down Expand Up @@ -492,12 +496,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of "
"seconds")
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port,
" ".join(map(repr, fields)))
Expand Down Expand Up @@ -527,11 +532,12 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of "
"seconds")
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))
12 changes: 8 additions & 4 deletions neo4j/_sync/io/_bolt3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,11 +229,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port, " ".join(map(repr, fields)))
if query.upper() == u"COMMIT":
Expand Down Expand Up @@ -281,11 +283,13 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))

Expand Down
30 changes: 18 additions & 12 deletions neo4j/_sync/io/_bolt4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,11 +181,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port, " ".join(map(repr, fields)))
if query.upper() == u"COMMIT":
Expand Down Expand Up @@ -232,11 +234,13 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))

Expand Down Expand Up @@ -492,12 +496,13 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of "
"seconds")
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
fields = (query, parameters, extra)
log.debug("[#%04X] C: RUN %s", self.local_port,
" ".join(map(repr, fields)))
Expand Down Expand Up @@ -527,11 +532,12 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
extra["tx_metadata"] = dict(metadata)
except TypeError:
raise TypeError("Metadata must be coercible to a dict")
if timeout:
if timeout is not None:
try:
extra["tx_timeout"] = int(1000 * timeout)
extra["tx_timeout"] = int(1000 * float(timeout))
except TypeError:
raise TypeError("Timeout must be specified as a number of "
"seconds")
raise TypeError("Timeout must be specified as a number of seconds")
if extra["tx_timeout"] < 0:
raise ValueError("Timeout must be a positive number or 0.")
log.debug("[#%04X] C: BEGIN %r", self.local_port, extra)
self._append(b"\x11", (extra,), Response(self, "begin", **handlers))
8 changes: 5 additions & 3 deletions neo4j/work/query.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ class Query:
:param metadata: metadata attached to the query.
:type metadata: dict
:param timeout: seconds.
:type timeout: int
:type timeout: float or None
"""
def __init__(self, text, metadata=None, timeout=None):
self.text = text
Expand Down Expand Up @@ -59,8 +59,10 @@ def count_people_tx(tx):
Transactions that execute longer than the configured timeout will be terminated by the database.
This functionality allows to limit query/transaction execution time.
Specified timeout overrides the default timeout configured in the database using ``dbms.transaction.timeout`` setting.
Value should not represent a duration of zero or negative duration.
:type timeout: int
Value should not represent a negative duration.
A zero duration will make the transaction execute indefinitely.
None will use the default timeout configured in the database.
:type timeout: float or None
"""

def wrapper(f):
Expand Down
53 changes: 37 additions & 16 deletions testkitbackend/_async/backend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
dumps,
loads,
)
from pathlib import Path
import traceback

from neo4j._exceptions import BoltError
Expand All @@ -42,6 +43,10 @@
from ..backend import Request


TESTKIT_BACKEND_PATH = Path(__file__).absolute().resolve().parents[1]
DRIVER_PATH = TESTKIT_BACKEND_PATH.parent / "neo4j"


class AsyncBackend:
def __init__(self, rd, wr):
self._rd = rd
Expand Down Expand Up @@ -81,6 +86,30 @@ async def process_request(self):
request = request + line
return False

@staticmethod
def _exc_stems_from_driver(exc):
stack = traceback.extract_tb(exc.__traceback__)
for frame in stack[-1:1:-1]:
p = Path(frame.filename)
if TESTKIT_BACKEND_PATH in p.parents:
return False
if DRIVER_PATH in p.parents:
return True

async def _handle_driver_exc(self, exc):
log.debug(traceback.format_exc())
if isinstance(exc, Neo4jError):
msg = "" if exc.message is None else str(exc.message)
else:
msg = str(exc.args[0]) if exc.args else ""

key = self.next_key()
self.errors[key] = exc
payload = {"id": key, "errorType": str(type(exc)), "msg": msg}
if isinstance(exc, Neo4jError):
payload["code"] = exc.code
await self.send_response("DriverError", payload)

async def _process(self, request):
""" Process a received request by retrieving handler that
corresponds to the request name.
Expand All @@ -104,24 +133,16 @@ async def _process(self, request):
)
except (Neo4jError, DriverError, UnsupportedServerProduct,
BoltError) as e:
log.debug(traceback.format_exc())
if isinstance(e, Neo4jError):
msg = "" if e.message is None else str(e.message)
else:
msg = str(e.args[0]) if e.args else ""

key = self.next_key()
self.errors[key] = e
payload = {"id": key, "errorType": str(type(e)), "msg": msg}
if isinstance(e, Neo4jError):
payload["code"] = e.code
await self.send_response("DriverError", payload)
await self._handle_driver_exc(e)
except requests.FrontendError as e:
await self.send_response("FrontendError", {"msg": str(e)})
except Exception:
tb = traceback.format_exc()
log.error(tb)
await self.send_response("BackendError", {"msg": tb})
except Exception as e:
if self._exc_stems_from_driver(e):
await self._handle_driver_exc(e)
else:
tb = traceback.format_exc()
log.error(tb)
await self.send_response("BackendError", {"msg": tb})

async def send_response(self, name, data):
""" Sends a response to backend.
Expand Down
8 changes: 4 additions & 4 deletions testkitbackend/_async/requests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,8 +253,8 @@ async def SessionClose(backend, data):
async def SessionBeginTransaction(backend, data):
key = data["sessionId"]
session = backend.sessions[key].session
metadata, timeout = fromtestkit.to_meta_and_timeout(data)
tx = await session.begin_transaction(metadata=metadata, timeout=timeout)
tx_kwargs = fromtestkit.to_tx_kwargs(data)
tx = await session.begin_transaction(**tx_kwargs)
key = backend.next_key()
backend.transactions[key] = tx
await backend.send_response("Transaction", {"id": key})
Expand All @@ -272,9 +272,9 @@ async def transactionFunc(backend, data, is_read):
key = data["sessionId"]
session_tracker = backend.sessions[key]
session = session_tracker.session
metadata, timeout = fromtestkit.to_meta_and_timeout(data)
tx_kwargs = fromtestkit.to_tx_kwargs(data)

@neo4j.unit_of_work(metadata=metadata, timeout=timeout)
@neo4j.unit_of_work(**tx_kwargs)
async def func(tx):
txkey = backend.next_key()
backend.transactions[txkey] = tx
Expand Down
Loading