From 92b8c012cf96b981c2230d7715c66880c45ce125 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 24 Oct 2026 11:03:21 +0900 Subject: [PATCH] Python functions callable from SQL and openCypher, and record event callbacks (#19) Ported from humemai/arcadedb-embedded-python #391, with the later fixes to the same files (#438 JPype race with Ctrl-C is a KeyboardInterrupt, #454 proxy classes defined once under a lock): - Database.register_function() / unregister_function(), callable from SQL and as py. from openCypher - Database.on() / off() record event callbacks (before_* may veto a write) - Document.save() and delete() raise ArcadeDBError instead of the raw Java exception - new Java glue: PythonFunctions.java and PythonRecordEvents.java (compiled into the existing arcadedb-python-bridge.jar by the build that already compiles src/java) - example 28 Based on port/runtime-fixes (#27). Co-Authored-By: Claude Opus 5.5 --- .../28_python_functions_and_record_events.py | 157 +++++ examples/README.md | 16 + src/arcadedb_embedded/__init__.py | 4 + src/arcadedb_embedded/callbacks.py | 404 +++++++++++ src/arcadedb_embedded/core.py | 147 +++- src/arcadedb_embedded/graph.py | 28 +- src/arcadedb_embedded/results.py | 8 +- .../com/arcadedb/python/PythonFunctions.java | 161 +++++ .../arcadedb/python/PythonRecordEvents.java | 117 ++++ tests/conftest.py | 2 + tests/test_open_read_only.py | 7 +- tests/test_python_callbacks.py | 639 ++++++++++++++++++ tests/test_python_function_interrupt.py | 84 +++ 13 files changed, 1763 insertions(+), 11 deletions(-) create mode 100644 examples/28_python_functions_and_record_events.py create mode 100644 src/arcadedb_embedded/callbacks.py create mode 100644 src/java/com/arcadedb/python/PythonFunctions.java create mode 100644 src/java/com/arcadedb/python/PythonRecordEvents.java create mode 100644 tests/test_python_callbacks.py create mode 100644 tests/test_python_function_interrupt.py diff --git a/examples/28_python_functions_and_record_events.py b/examples/28_python_functions_and_record_events.py new file mode 100644 index 0000000..30fd35a --- /dev/null +++ b/examples/28_python_functions_and_record_events.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""Example 28: Python functions in SQL and openCypher, and record event callbacks. + +Two things no query language can express: +- db.register_function() makes a Python function callable from SQL and openCypher +- db.on() runs a Python callback when a record is created, updated, or deleted + +Workflow covered: +- register `slugify` and call it in SELECT and WHERE (SQL), and as `py.slugify` from openCypher +- see an exception in the function arrive as a query error +- measure the cost of one call (a plain SQL expression against the same expression in Python) +- keep an audit trail with after_create / after_update / after_delete callbacks +- veto a write with a before_create callback that returns False +- remove a callback and unregister a function + +Notes: +- No data download; a few seconds end to end. +- The engine registers functions for the whole process, and a Database removes the ones + it registered when it closes. +""" + +from __future__ import annotations + +import argparse +import re +import shutil +import time + +import arcadedb_embedded as arcadedb + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("--rows", type=int, default=20_000, help="rows for the cost check") + p.add_argument("--db-path", default="./my_test_databases/python_functions_events") + return p.parse_args() + + +def slugify(text): + """NULL in, NULL out: the engine passes SQL NULL as None.""" + if text is None: + return None + return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + + +def show_functions(db) -> None: + print("\n== Python functions ==") + with db.transaction(): + for name in ("Ann Lee", "Bob O'Ray", None): + db.command("sql", "INSERT INTO Person SET name = ?", name) + db.command("sql", "CREATE VERTEX City SET name = 'New York'") + + db.register_function("slugify", slugify) + # openCypher accepts a namespaced name for functions it does not know + db.register_function("py.slugify", slugify) + + rows = db.query("sql", "SELECT name, slugify(name) AS slug FROM Person").to_list() + for row in rows: + print(f" SELECT {row['name']!r:14} -> {row['slug']!r}") + + hits = db.query("sql", "SELECT name FROM Person WHERE slugify(name) = 'ann-lee'") + print(" WHERE matches:", [row["name"] for row in hits.to_list()]) + + cypher = "MATCH (c:City) RETURN py.slugify(c.name) AS slug" + print(" openCypher:", db.query("opencypher", cypher).to_list()) + + def explode(value): + raise ValueError(f"cannot handle {value!r}") + + db.register_function("explode", explode) + try: + db.query("sql", "SELECT explode(name) FROM Person").to_list() + except arcadedb.ArcadeDBError as e: + print(" an exception in the function is a query error:\n ", e) + + db.unregister_function("explode") + + +def measure_call_cost(db, rows: int) -> None: + print("\n== Cost of one call ==") + with db.transaction(): + for i in range(rows): + db.command("sql", "INSERT INTO Number SET n = ?", i) + db.register_function("plus_one", lambda n: n + 1) + + def best_of(statement: str, runs: int = 5) -> float: + best = float("inf") + for _ in range(runs): + start = time.perf_counter() + db.query("sql", statement).to_list() + best = min(best, time.perf_counter() - start) + return best + + with_function = best_of("SELECT plus_one(n) AS m FROM Number") + plain_sql = best_of("SELECT n + 1 AS m FROM Number") + per_call_us = (with_function - plain_sql) / rows * 1e6 + print( + f" {rows} rows: plus_one(n) {with_function * 1e3:.0f} ms, n + 1 {plain_sql * 1e3:.0f} ms" + ) + print( + f" about {per_call_us:.1f} microseconds per Python call (one JNI round trip each)" + ) + db.unregister_function("plus_one") + + +def show_events(db) -> None: + print("\n== Record events ==") + audit = [] + + def record_event(event): + def callback(record): + audit.append((event, record.get_type_name(), record.get("name"))) + + return callback + + handles = [ + db.on(event, record_event(event), "Account") + for event in ("after_create", "after_update", "after_delete") + ] + + def no_blank_names(record): + return bool(record.get("name")) # False vetoes the write + + veto = db.on("before_create", no_blank_names, "Account") + + with db.transaction(): + db.command("sql", "INSERT INTO Account SET name = 'acme'") + db.command("sql", "INSERT INTO Account SET name = ''") # vetoed, skipped + db.command("sql", "UPDATE Account SET name = 'acme inc' WHERE name = 'acme'") + db.command("sql", "DELETE FROM Account WHERE name = 'acme inc'") + + for entry in audit: + print(" audit:", entry) + print(" accounts left:", db.count_type("Account")) + + for handle in handles + [veto]: + handle.remove() + with db.transaction(): + db.command("sql", "INSERT INTO Account SET name = 'after removal'") + print(" audit entries after remove():", len(audit)) + + +def main() -> None: + args = parse_args() + shutil.rmtree(args.db_path, ignore_errors=True) + with arcadedb.create_database(args.db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE Person") + db.command("sql", "CREATE DOCUMENT TYPE Number") + db.command("sql", "CREATE DOCUMENT TYPE Account") + db.command("sql", "CREATE VERTEX TYPE City") + show_functions(db) + measure_call_cost(db, args.rows) + show_events(db) + + +if __name__ == "__main__": + main() diff --git a/examples/README.md b/examples/README.md index e167306..239e8e9 100644 --- a/examples/README.md +++ b/examples/README.md @@ -391,6 +391,22 @@ half-applied after an interruption --- +### 🧩 [28_python_functions_and_record_events.py](./28_python_functions_and_record_events.py) +**Python Functions | Record Events | SQL + openCypher** + +Two things no query language can express: +- `db.register_function()` makes a Python `slugify` callable from SQL (SELECT and WHERE) and, + as `py.slugify`, from openCypher; an exception in the function is a query error +- measures the cost of one call on your machine against the same SQL expression +- `db.on()` keeps an audit trail with `after_create` / `after_update` / `after_delete` + callbacks and vetoes a write with a `before_create` callback that returns `False` +- no data download; flags: `--rows` (default 20,000) and `--db-path` + +**Learn:** When a Python function or callback is the right tool, what it costs, and how +errors, threads, and unregistering behave + +--- + ## 💡 Tips - **Run from examples/ directory** - Always execute examples from `examples/` for correct file paths diff --git a/src/arcadedb_embedded/__init__.py b/src/arcadedb_embedded/__init__.py index 83e575a..0cdf495 100644 --- a/src/arcadedb_embedded/__init__.py +++ b/src/arcadedb_embedded/__init__.py @@ -25,6 +25,9 @@ # Import async execution from .async_executor import AsyncExecutor +# Import record event handle +from .callbacks import EventHandle + # Import core database classes from .core import ( Database, @@ -84,6 +87,7 @@ # Core classes "Database", "DatabaseFactory", + "EventHandle", # Record wrappers "Document", "Vertex", diff --git a/src/arcadedb_embedded/callbacks.py b/src/arcadedb_embedded/callbacks.py new file mode 100644 index 0000000..62274ac --- /dev/null +++ b/src/arcadedb_embedded/callbacks.py @@ -0,0 +1,404 @@ +""" +ArcadeDB Python Bindings - Python functions in SQL and record event callbacks + +``Database.register_function()`` makes a Python callable usable as a SQL (and openCypher) +function. ``Database.on()`` runs a Python callback when a record is created, updated, or +deleted. Both ride the bridge jar (``PythonFunctions``, ``PythonRecordEvents``), which +implements the engine's interfaces and forwards each call to a JPype proxy defined here. +""" + +import inspect +import re +import threading +import traceback +import weakref +from typing import Callable, Optional, Tuple + +import jpype + +from ._logging import get_logger +from .exceptions import ArcadeDBError +from .type_conversion import convert_java_to_python, convert_python_to_java + +_LOGGER = get_logger(__name__) + +EVENTS = ( + "before_create", + "after_create", + "before_update", + "after_update", + "before_delete", + "after_delete", +) + +_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$") +_UNBOUNDED = 2**31 - 1 # Integer.MAX_VALUE: the engine's "any number of arguments" + + +def _bridge_class(name: str): + """Load a bridge class, or raise an ArcadeDBError that says what is missing.""" + try: + return jpype.JClass(name) + except Exception as e: + raise ArcadeDBError( + f"Bridge class {name} is unavailable: this feature needs " + "arcadedb-python-bridge.jar from the same wheel as the Python package" + ) from e + + +# A KeyboardInterrupt raised inside a Python function or callback cannot cross the engine as it +# is: it ends the statement as a Java exception. It is remembered per thread, and the place that +# turns the statement's Java exception into an ArcadeDBError raises the KeyboardInterrupt instead +# (chained from that ArcadeDBError), so Ctrl-C reaches the caller as KeyboardInterrupt whether the +# signal landed in the function, in the engine, or between the two. +_PENDING = threading.local() + + +def _is_jpype_interrupt_race(error: BaseException) -> bool: + """True for the ``RuntimeError`` JPype raises when it loses its race with Ctrl-C. + + JPype's SIGINT handler (``interrupt=False``) interrupts the Java thread first and tells + Python afterwards (test_sigint.py, jpype-project/jpype#1496). JPype's own Java + call, while it converts a Java exception to a Python one, then fails with + ``InterruptedException: sleep interrupted``, and the call ends in + ``RuntimeError: Fatal error occurred``. It means Ctrl-C: the signal reached the statement + while the engine handled the failure of a Python function (a KeyboardInterrupt raised on + entering the callback arrives as ``org.jpype.PyExceptionProxy``, one raised in the + function as ``KeyboardInterrupt``). + """ + return isinstance(error, RuntimeError) and "Fatal error occurred" in str(error) + + +def _raise_to_java(error: BaseException): + """Hand a Python exception to the engine as a Java RuntimeException. + + The text names the Python exception type, so the query error says what went wrong. The + Python traceback goes to the debug log. Nothing escapes into the JVM as a Python exception. + """ + _LOGGER.debug( + "Python callback raised:\n%s", + "".join(traceback.format_exception(type(error), error, error.__traceback__)), + ) + if isinstance(error, KeyboardInterrupt): + _PENDING.interrupt = error + elif _is_jpype_interrupt_race(error): + _PENDING.interrupt = KeyboardInterrupt( + "interrupted while a Python function was called" + ) + _PENDING.interrupt.__cause__ = error + raise jpype.JClass("java.lang.RuntimeException")(f"{type(error).__name__}: {error}") + + +def clear_pending_interrupt() -> None: + """Forget a KeyboardInterrupt left by an earlier statement (called when a statement starts).""" + _PENDING.interrupt = None + + +def _escaped_python_exception(exc: BaseException) -> bool: + """True when a Java exception chain holds JPype's proxy for a Python exception that left a callback. + + Only an exception that arrives outside the callback's own ``try`` can do that, and that is + in practice a KeyboardInterrupt raised by the signal handler on entering the callback. + """ + try: + cause = exc + for _ in range(16): + if cause is None: + return False + if "PyExceptionProxy" in str(cause.getClass().getName()): + return True + cause = cause.getCause() + except Exception: # nosec B110 - not a Java exception: nothing escaped + pass + return False + + +def raise_pending_interrupt(error: BaseException, java_exc: BaseException) -> None: + """Raise the KeyboardInterrupt a Python callback left, chained from ``error``, if there is one.""" + interrupt = getattr(_PENDING, "interrupt", None) + _PENDING.interrupt = None + if interrupt is None and ( + _escaped_python_exception(java_exc) or _is_jpype_interrupt_race(java_exc) + ): + interrupt = KeyboardInterrupt("interrupted while a Python function was called") + if interrupt is not None: + raise interrupt from error + + +def _arity(fn: Callable) -> Tuple[int, int]: + """The (min, max) number of positional arguments ``fn`` takes; max is unbounded for ``*args``.""" + try: + parameters = inspect.signature(fn).parameters.values() + except (TypeError, ValueError): + return 0, _UNBOUNDED + minimum = maximum = 0 + for parameter in parameters: + if parameter.kind in ( + inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + ): + maximum += 1 + if parameter.default is inspect.Parameter.empty: + minimum += 1 + elif parameter.kind is inspect.Parameter.VAR_POSITIONAL: + maximum = _UNBOUNDED + elif ( + parameter.kind is inspect.Parameter.KEYWORD_ONLY + and parameter.default is inspect.Parameter.empty + ): + raise ArcadeDBError( + f"Function parameter '{parameter.name}' is keyword-only and has no " + "default: SQL passes positional arguments only" + ) + return minimum, maximum + + +# The proxy classes are defined once, on first use (the JVM must be running, so not at import): +# a @JImplements class defined on every registration leaked about 6 KB of Python heap per +# registration, which neither the engine's unregister nor gc gave back. State is per instance. +_PROXY_CLASSES = {} +# defined under a lock: on a free-threaded CPython (no GIL) two threads that register their first function or event together would +# each define a class, and the second definition leaks like the one this cache exists to avoid +_PROXY_LOCK = threading.Lock() + + +def _function_proxy_class(): + cls = _PROXY_CLASSES.get("function") + if cls is None: + with _PROXY_LOCK: + cls = _PROXY_CLASSES.get("function") + if cls is None: + + @jpype.JImplements( + _bridge_class("com.arcadedb.python.PythonFunctions$Callback") + ) + class _PythonFunction: + def __init__(self, fn, database_ref): + self._fn = fn + self._database_ref = database_ref + + @jpype.JOverride + def call(self, args): + from .core import _wrap_java_record + + try: + values = [] + for arg in args: + value = convert_java_to_python(arg) + values.append( + _wrap_java_record(value, self._database_ref()) + ) + result = convert_python_to_java(self._fn(*values)) + if result is not None: + # Convert to a Java Object here, inside the try: left to JPype's return + # conversion an unconvertible value (a numpy array, an int past 2**63) + # reaches the engine as an opaque org.jpype.PyExceptionProxy. + result = jpype.JObject(result, jpype.JObject) + return result + except BaseException as e: # pylint: disable=broad-except + _raise_to_java(e) + + cls = _PROXY_CLASSES["function"] = _PythonFunction + return cls + + +def _event_proxy_class(): + cls = _PROXY_CLASSES.get("event") + if cls is None: + with _PROXY_LOCK: + cls = _PROXY_CLASSES.get("event") + if cls is None: + + @jpype.JImplements( + _bridge_class("com.arcadedb.python.PythonRecordEvents$Callback") + ) + class _PythonListener: + def __init__(self, fn, is_before, database_ref): + self._fn = fn + self._is_before = is_before + self._database_ref = database_ref + + @jpype.JOverride + def call(self, record): + from .core import _wrap_java_record + + try: + result = self._fn( + _wrap_java_record(record, self._database_ref()) + ) + except BaseException as e: # pylint: disable=broad-except + _raise_to_java(e) + return not (self._is_before and result is False) + + cls = _PROXY_CLASSES["event"] = _PythonListener + return cls + + +def _function_callback(fn: Callable, database): + """A PythonFunctions.Callback proxy: Java arguments in, converted result out.""" + # weak: the engine holds this proxy, so a strong Database here would never be released + return _function_proxy_class()(fn, weakref.ref(database)) + + +def _event_callback(event: str, fn: Callable, database): + """A PythonRecordEvents.Callback proxy: the record wrapped, a before event may veto.""" + # weak: the engine holds this proxy, so a strong Database here would never be released + return _event_proxy_class()(fn, event.startswith("before_"), weakref.ref(database)) + + +class FunctionRegistration: + """Python functions registered through one ``Database``. + + The engine registers functions for the whole process, so each ``Database`` keeps the ones + it registered to remove them when it is closed. + """ + + def __init__(self, database): + # Weak: the Database holds this registry, and a strong reference back would make a + # cycle that keeps the Database (and its engine instance) alive until a gc pass + # instead of releasing it with its last result (tests/test_results_after_close.py). + self._database_ref = weakref.ref(database) + self._handles = {} + + @property + def _database(self): + return self._database_ref() + + def register( + self, + name: str, + fn: Callable, + min_args: Optional[int], + max_args: Optional[int], + deterministic: bool, + replace: bool, + ) -> None: + if not isinstance(name, str) or not _NAME_RE.match(name): + raise ValueError( + f"Invalid function name {name!r}: use letters, digits and underscores " + "(not starting with a digit), optionally joined by dots like 'py.slugify'" + ) + if not callable(fn): + raise TypeError("fn must be callable") + found_min, found_max = _arity(fn) + min_args = found_min if min_args is None else min_args + max_args = found_max if max_args is None else max_args + if min_args < 0 or max_args < min_args: + raise ValueError( + f"Invalid argument counts: min_args={min_args}, max_args={max_args}" + ) + + functions = _bridge_class("com.arcadedb.python.PythonFunctions") + try: + if functions.isTakenByEngine(name): + raise ArcadeDBError( + f"Cannot register '{name}': it is already a function of the engine" + ) + if functions.isPythonFunction(name) and not replace: + raise ArcadeDBError( + f"A Python function '{name}' is already registered " + "(pass replace=True to replace it)" + ) + callback = _function_callback(fn, self._database) + handle = functions.register( + name, callback, min_args, max_args, bool(deterministic) + ) + except ArcadeDBError: + raise + except Exception as e: + raise ArcadeDBError(f"Failed to register function '{name}': {e}") from e + # keep the proxy alive for as long as the engine can call it + self._handles[name.lower()] = (handle, callback) + + def unregister(self, name: str) -> None: + entry = self._handles.pop(str(name).lower(), None) + if entry is None: + raise ArcadeDBError( + f"No Python function '{name}' is registered through this database" + ) + _bridge_class("com.arcadedb.python.PythonFunctions").unregisterHandle(entry[0]) + + def clear(self) -> None: + """Remove every function this database registered (called when it closes).""" + handles, self._handles = self._handles, {} + try: + functions = jpype.JClass("com.arcadedb.python.PythonFunctions") + for handle, _callback in handles.values(): + functions.unregisterHandle(handle) + except Exception: # nosec B110 - the JVM may already be shutting down + pass + + +class EventHandle: + """A registered record event callback; ``remove()`` stops it. + + Returned by ``Database.on()``. It is also a context manager that removes the callback on + exit. + """ + + def __init__( + self, events, event: str, type_name: Optional[str], java_handle, proxy + ): + self._events = events + self._java_handle = java_handle + self._proxy = proxy + self.event = event + self.type_name = type_name + self._active = True + + @property + def active(self) -> bool: + """Whether the callback is still registered.""" + return self._active + + def remove(self) -> None: + """Unregister the callback. Calling it again does nothing.""" + if not self._active: + return + self._active = False + try: + _bridge_class("com.arcadedb.python.PythonRecordEvents").unregister( + self._events, self.event, self._java_handle + ) + except Exception as e: + raise ArcadeDBError(f"Failed to remove '{self.event}' callback: {e}") from e + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.remove() + + def __repr__(self) -> str: + scope = f" on {self.type_name}" if self.type_name else "" + state = "active" if self._active else "removed" + return f"" + + +def register_event(database, event: str, fn: Callable, type_name: Optional[str]): + """Register ``fn`` for ``event`` on the database, or on one type, and return its handle.""" + if event not in EVENTS: + raise ValueError(f"Invalid event {event!r}. Must be one of: {list(EVENTS)}") + if not callable(fn): + raise TypeError("fn must be callable") + java_db = database.get_java_database() + try: + if type_name is None: + events = java_db.getEvents() + else: + schema = java_db.getSchema() + if not schema.existsType(type_name): + raise ArcadeDBError(f"Type '{type_name}' does not exist") + events = schema.getType(type_name).getEvents() + proxy = _event_callback(event, fn, database) + java_handle = _bridge_class("com.arcadedb.python.PythonRecordEvents").register( + events, event, proxy + ) + except ArcadeDBError: + raise + except Exception as e: + raise ArcadeDBError(f"Failed to register '{event}' callback: {e}") from e + return EventHandle(events, event, type_name, java_handle, proxy) + + +__all__ = ["EVENTS", "EventHandle", "FunctionRegistration", "register_event"] diff --git a/src/arcadedb_embedded/core.py b/src/arcadedb_embedded/core.py index 2a85037..63f8a43 100644 --- a/src/arcadedb_embedded/core.py +++ b/src/arcadedb_embedded/core.py @@ -9,11 +9,18 @@ from datetime import date from decimal import Decimal from os import PathLike -from typing import Any, List, Optional +from typing import Any, Callable, List, Optional import jpype from ._logging import get_logger +from .callbacks import ( + EventHandle, + FunctionRegistration, + clear_pending_interrupt, + raise_pending_interrupt, + register_event, +) from .exceptions import ArcadeDBError, _retry_verdict from .graph import Document, Edge, Vertex from .graph_batch import GraphBatch @@ -264,6 +271,7 @@ def __init__(self, java_database): self._java_db = java_database self._closed = False self._async_executors = [] + self._functions = FunctionRegistration(self) def __enter__(self): return self @@ -371,6 +379,7 @@ def query(self, language: str, command: str, *args) -> ResultSet: is the positional list itself; ``None`` binds as null. """ self._check_not_closed() + clear_pending_interrupt() try: if args: java_result = _through_bridge( @@ -384,7 +393,10 @@ def query(self, language: str, command: str, *args) -> ResultSet: java_result = self._java_db.query(language, command) return ResultSet(java_result, self) except Exception as e: - raise ArcadeDBError(f"Query failed: {e}") from e + error = ArcadeDBError(f"Query failed: {e}") + error.__cause__ = e + raise_pending_interrupt(error, e) + raise error from e def command(self, language: str, command: str, *args) -> Optional[ResultSet]: """Execute a command (non-idempotent operation). @@ -392,6 +404,7 @@ def command(self, language: str, command: str, *args) -> Optional[ResultSet]: Parameters bind as in :meth:`query`. """ self._check_not_closed() + clear_pending_interrupt() try: if args: java_result = _through_bridge( @@ -408,7 +421,10 @@ def command(self, language: str, command: str, *args) -> Optional[ResultSet]: return ResultSet(java_result, self) return None except Exception as e: - raise ArcadeDBError(f"Command failed: {e}") from e + error = ArcadeDBError(f"Command failed: {e}") + error.__cause__ = e + raise_pending_interrupt(error, e) + raise error from e def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): """ @@ -1000,6 +1016,7 @@ def close(self): """Close the database.""" if not self._closed and self._java_db is not None: async_close_error = None + self._functions.clear() try: async_close_error = self._close_async_executors() self._java_db.close() @@ -1521,6 +1538,130 @@ def set_auto_transaction(self, enabled: bool): except Exception as e: raise ArcadeDBError(f"Failed to set auto-transaction: {e}") from e + def register_function( + self, + name: str, + fn: Callable, + *, + min_args: Optional[int] = None, + max_args: Optional[int] = None, + deterministic: bool = False, + replace: bool = False, + ) -> None: + """ + Register a Python function that SQL and openCypher statements can call. + + After ``db.register_function("slugify", fn)``, ``SELECT slugify(name) FROM Person`` + and ``MATCH (p:Person) RETURN slugify(p.name)`` call ``fn`` once per row. + Arguments reach ``fn`` as Python values (SQL ``NULL`` is ``None``) and the return + value is converted back with the bindings' type conversion, so ``fn`` may return + ``None``, a number, a string, a bool, a ``Decimal``, a ``datetime``, a list, or a + dict. An exception raised in ``fn`` fails the statement with an ``ArcadeDBError`` + that names the function and the Python exception. A ``KeyboardInterrupt`` (Ctrl-C) + is the one exception to that: the statement is aborted and the caller gets a + ``KeyboardInterrupt``, chained from the ``ArcadeDBError``, wherever the signal + landed. A numeric argument costs about 30 bytes of Python memory per call that + JPype 1.7.1 does not give back (see the Python functions guide). + + The engine registers functions for the whole process, not for one database: every + database in this process sees the function, and a function stays registered until + :meth:`unregister_function` or until this ``Database`` is closed. Register before + running concurrent statements: the engine's function table is not safe to change + while another thread resolves a name. The engine may call ``fn`` on its own + threads; the GIL serializes the calls, so ``fn`` does not run in parallel. An + aggregate (one result per group) cannot be written this way. + + Args: + name: Function name; letters, digits and underscores, starting with a letter + or underscore. SQL function names are case-insensitive. A name that is + already an engine function (``abs``, ``count``, ``toUpper``, ...) is + refused. + fn: The callable. It takes positional arguments only. + min_args: Fewest arguments a call may pass. Default: the parameters of ``fn`` + without a default. + max_args: Most arguments a call may pass. Default: all positional parameters + of ``fn`` (unlimited with ``*args``). + deterministic: True only if ``fn`` returns the same result for the same + arguments and reads nothing else. The planner may then fold a call over + constants at plan time and cache plans that contain it. + replace: Replace a Python function already registered under this name. + + Raises: + ValueError: If the name or an argument count is invalid + ArcadeDBError: If the name is an engine function, or a Python function with + this name exists and ``replace`` is false + + Example: + >>> db.register_function("slugify", lambda s: s.lower().replace(" ", "-")) + >>> db.query("sql", "SELECT slugify('Hello World') AS s").first().get("s") + 'hello-world' + """ + self._check_not_closed() + self._functions.register(name, fn, min_args, max_args, deterministic, replace) + + def unregister_function(self, name: str) -> None: + """ + Remove a Python function registered through this database. + + A statement that calls it afterwards fails with an unknown-function error. + + Args: + name: The name given to :meth:`register_function` + + Raises: + ArcadeDBError: If this database did not register a function with this name + """ + self._check_not_closed() + self._functions.unregister(name) + + def on( + self, + event: str, + fn: Callable, + type_name: Optional[str] = None, + ) -> EventHandle: + """ + Call a Python function when a record is created, updated, or deleted. + + ``fn(record)`` receives the record as a :class:`Vertex`, :class:`Edge`, or + :class:`Document`. It runs inside the operation that fired the event, on the + thread that runs it (the engine may use its own threads, for example for async + writes; the GIL serializes the calls). A callback should be quick and should not + start work that waits on another thread's transaction. + + Events: ``before_create``, ``after_create``, ``before_update``, ``after_update``, + ``before_delete``, ``after_delete``. A ``before_*`` callback that returns + ``False`` vetoes the change: the record is not written (a ``save()`` or a + statement skips it silently, as for the engine's own listeners). Any other return + value allows it, and the return value of an ``after_*`` callback is ignored. + + An exception raised in a callback is not swallowed: it fails the operation that + fired the event with an ``ArcadeDBError`` that names the event and the Python + exception, and the surrounding transaction does not commit. A ``before_*`` event + fires before the record is written, an ``after_*`` event right after the write + and before the commit, so a rolled-back transaction has already fired its + ``after_*`` callbacks. + + Args: + event: One of the event names above + fn: The callback, called with one argument: the record + type_name: Only records of exactly this type (not its subtypes). Default: every + record of the database. + + Returns: + An :class:`EventHandle`; ``handle.remove()`` stops the callback + + Raises: + ValueError: If ``event`` is not valid + ArcadeDBError: If ``type_name`` does not exist + + Example: + >>> handle = db.on("after_create", lambda rec: print(rec.get_rid()), "Person") + >>> handle.remove() + """ + self._check_not_closed() + return register_event(self, event, fn, type_name) + def async_executor(self): """ Get async executor for parallel operations. diff --git a/src/arcadedb_embedded/graph.py b/src/arcadedb_embedded/graph.py index 1c2f257..4641788 100644 --- a/src/arcadedb_embedded/graph.py +++ b/src/arcadedb_embedded/graph.py @@ -8,6 +8,7 @@ import jpype +from .callbacks import raise_pending_interrupt from .exceptions import ArcadeDBError from .type_conversion import convert_java_to_python, convert_python_to_java @@ -123,8 +124,19 @@ def set(self, name: str, value: Any) -> "Document": return self def save(self) -> "Document": - """Save the document.""" - self._java_document.save() + """Save the document. + + Raises: + ArcadeDBError: If the engine refuses the write (no transaction, a duplicate key, + a read-only database, a ``before_*`` callback that raised). + """ + try: + self._java_document.save() + except jpype.JException as e: + error = ArcadeDBError(f"Failed to save record: {e}") + error.__cause__ = e + raise_pending_interrupt(error, e) + raise error from e return self def delete(self): @@ -132,8 +144,18 @@ def delete(self): A query row is a ``Result``, not a record: reach the record with ``result.get_element()`` (or ``db.lookup_by_rid()``) and delete that. + + Raises: + ArcadeDBError: If the engine refuses the delete (no transaction, a read-only + database, a ``before_delete`` callback that raised). """ - self._java_document.delete() + try: + self._java_document.delete() + except jpype.JException as e: + error = ArcadeDBError(f"Failed to delete record: {e}") + error.__cause__ = e + raise_pending_interrupt(error, e) + raise error from e def modify(self) -> "Document": """Get mutable version for updates.""" diff --git a/src/arcadedb_embedded/results.py b/src/arcadedb_embedded/results.py index a8f78f8..b210cc8 100644 --- a/src/arcadedb_embedded/results.py +++ b/src/arcadedb_embedded/results.py @@ -13,6 +13,7 @@ from jpype import JException from ._logging import get_logger +from .callbacks import raise_pending_interrupt from .exceptions import ArcadeDBError from .graph import Document, Edge, Vertex from .type_conversion import convert_java_to_python @@ -91,7 +92,12 @@ def _read_error(exc: BaseException) -> ArcadeDBError: or any later row, after ``query()`` or ``command()`` returned (#173). Raise it ``from`` the Java exception, which ``str()`` then names as the cause. """ - return ArcadeDBError(f"Reading the result set failed: {exc}") + error = ArcadeDBError(f"Reading the result set failed: {exc}") + error.__cause__ = exc + raise_pending_interrupt( + error, exc + ) # raises KeyboardInterrupt when a Python function left one + return error def _java_class_name(value: Any) -> str: diff --git a/src/java/com/arcadedb/python/PythonFunctions.java b/src/java/com/arcadedb/python/PythonFunctions.java new file mode 100644 index 0000000..b8c26a5 --- /dev/null +++ b/src/java/com/arcadedb/python/PythonFunctions.java @@ -0,0 +1,161 @@ +/* + * Python-bindings bridge: Python functions callable from SQL and openCypher. + * + * The engine resolves a function name through the process-wide DefaultSQLFunctionFactory, for SQL and for openCypher + * alike (a name the Cypher executor does not know itself falls through to it). {@code Database.register_function()} + * wraps the Python callable in a {@link Callback} (a JPype proxy) and registers it here as a plain SQLFunction. The + * adapter owns what Python should not have to: the arity check, the wrapping of any failure of the callable into one + * CommandExecutionException that names the function, and a revoked flag, so a plan that still holds the function + * after unregister() fails with a clear error instead of calling into a callable the caller has given up. + * + * Compiled into arcadedb-python-bridge.jar during the wheel build (scripts/Dockerfile.build and + * scripts/build-native.sh) and consumed by Database.register_function() in the Python bindings. + */ +package com.arcadedb.python; + +import com.arcadedb.database.Identifiable; +import com.arcadedb.exception.CommandExecutionException; +import com.arcadedb.function.CypherBuiltinFunctions; +import com.arcadedb.function.CypherFunctionRegistry; +import com.arcadedb.function.FunctionRegistry; +import com.arcadedb.function.sql.DefaultSQLFunctionFactory; +import com.arcadedb.function.sql.SQLFunctionAbstract; +import com.arcadedb.query.sql.executor.CommandContext; + +import java.util.Locale; + +public final class PythonFunctions { + + /** The Python side of a function: the arguments as the engine passes them, the result as the engine takes it. */ + public interface Callback { + Object call(Object[] args) throws Exception; + } + + private static final class PythonSQLFunction extends SQLFunctionAbstract { + private final Callback callback; + private final int minArgs; + private final int maxArgs; + private final boolean deterministic; + private volatile boolean revoked; + + PythonSQLFunction(final String name, final Callback callback, final int minArgs, final int maxArgs, + final boolean deterministic) { + super(name); + this.callback = callback; + this.minArgs = minArgs; + this.maxArgs = maxArgs; + this.deterministic = deterministic; + } + + @Override + public int getMinArgs() { + return minArgs; + } + + @Override + public int getMaxArgs() { + return maxArgs; + } + + @Override + public boolean isDeterministic() { + return deterministic; + } + + @Override + public String getSyntax() { + return name + "(" + (maxArgs == 0 ? "" : "") + ") [Python]"; + } + + @Override + public Object execute(final Object self, final Identifiable currentRecord, final Object currentResult, + final Object[] params, final CommandContext context) { + if (revoked) + throw new CommandExecutionException("Python function '" + name + "' was unregistered"); + if (params.length < minArgs || params.length > maxArgs) + throw new CommandExecutionException("Python function '" + name + "' takes " + arity() + " argument(s), got " + params.length); + try { + return callback.call(params); + } catch (final Throwable t) { + throw new CommandExecutionException("Python function '" + name + "' failed: " + describe(t), t); + } + } + + private String arity() { + if (minArgs == maxArgs) + return String.valueOf(minArgs); + return maxArgs == Integer.MAX_VALUE ? "at least " + minArgs : minArgs + " to " + maxArgs; + } + } + + private PythonFunctions() { + } + + static String describe(final Throwable t) { + final String message = t.getMessage(); + return message == null || message.isEmpty() ? t.getClass().getName() : message; + } + + private static String key(final String name) { + return name.toLowerCase(Locale.ENGLISH); + } + + /** + * Whether {@code name} already resolves to something that is not a Python function: a built-in SQL function, a + * Cypher function, or one an application registered from Java. Such a name is never replaced. + */ + public static synchronized boolean isTakenByEngine(final String name) { + final String key = key(name); + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + if (factory.hasFunction(key)) + return !(factory.getFunctionInstance(key) instanceof PythonSQLFunction); + return CypherBuiltinFunctions.isBuiltin(key) || CypherFunctionRegistry.hasFunction(key) || FunctionRegistry.hasFunction(key); + } + + /** Whether {@code name} is currently a Python function. */ + public static synchronized boolean isPythonFunction(final String name) { + final String key = key(name); + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + return factory.hasFunction(key) && factory.getFunctionInstance(key) instanceof PythonSQLFunction; + } + + /** Registers (or replaces a Python function of the same name) for SQL and openCypher. Returns the handle. */ + public static synchronized Object register(final String name, final Callback callback, final int minArgs, + final int maxArgs, final boolean deterministic) { + if (isTakenByEngine(name)) + throw new IllegalArgumentException("'" + name + "' is already a function of the engine"); + unregister(name); + final PythonSQLFunction function = new PythonSQLFunction(key(name), callback, minArgs, maxArgs, deterministic); + DefaultSQLFunctionFactory.getInstance().register(function); + return function; + } + + /** + * Removes the function {@code handle} (as returned by {@link #register}) if it is still the one registered under its + * name, and revokes it. Returns whether it was registered. + */ + public static synchronized boolean unregisterHandle(final Object handle) { + if (!(handle instanceof PythonSQLFunction function)) + return false; + return remove(function.getName(), function); + } + + /** Removes whatever Python function is registered under {@code name}. Returns whether there was one. */ + public static synchronized boolean unregister(final String name) { + final String key = key(name); + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + if (!factory.hasFunction(key)) + return false; + return factory.getFunctionInstance(key) instanceof PythonSQLFunction function && remove(key, function); + } + + private static boolean remove(final String key, final PythonSQLFunction function) { + function.revoked = true; + final DefaultSQLFunctionFactory factory = DefaultSQLFunctionFactory.getInstance(); + if (!factory.hasFunction(key) || factory.getFunctionInstance(key) != function) + return false; + factory.unregister(key); + FunctionRegistry.unregister(key); + return true; + } +} diff --git a/src/java/com/arcadedb/python/PythonRecordEvents.java b/src/java/com/arcadedb/python/PythonRecordEvents.java new file mode 100644 index 0000000..e106c27 --- /dev/null +++ b/src/java/com/arcadedb/python/PythonRecordEvents.java @@ -0,0 +1,117 @@ +/* + * Python-bindings bridge: Python callbacks on record events. + * + * RecordEvents has eight register/unregister overloads, one listener interface per event. One adapter class implements + * the six write events (before and after create, update and delete) and forwards each to one {@link Callback}, so + * Python implements ONE interface instead of six, and the veto of a before event is the boolean the callback returns. + * A failure of the callback is wrapped in a DatabaseOperationException that names the event: the engine does not catch + * it, so it reaches the statement or save() that fired the event and the transaction does not commit. + * + * Compiled into arcadedb-python-bridge.jar during the wheel build (scripts/Dockerfile.build and + * scripts/build-native.sh) and consumed by Database.on() in the Python bindings. + */ +package com.arcadedb.python; + +import com.arcadedb.database.Record; +import com.arcadedb.database.RecordEvents; +import com.arcadedb.event.AfterRecordCreateListener; +import com.arcadedb.event.AfterRecordDeleteListener; +import com.arcadedb.event.AfterRecordUpdateListener; +import com.arcadedb.event.BeforeRecordCreateListener; +import com.arcadedb.event.BeforeRecordDeleteListener; +import com.arcadedb.event.BeforeRecordUpdateListener; +import com.arcadedb.exception.DatabaseOperationException; + +public final class PythonRecordEvents { + + /** The Python side of a listener: false from a before event vetoes the change, the result of an after event is ignored. */ + public interface Callback { + boolean call(Record record) throws Exception; + } + + private static final class Listener + implements BeforeRecordCreateListener, BeforeRecordUpdateListener, BeforeRecordDeleteListener, AfterRecordCreateListener, + AfterRecordUpdateListener, AfterRecordDeleteListener { + private final String event; + private final Callback callback; + + Listener(final String event, final Callback callback) { + this.event = event; + this.callback = callback; + } + + private boolean dispatch(final Record record) { + try { + return callback.call(record); + } catch (final Throwable t) { + throw new DatabaseOperationException("Python '" + event + "' callback failed: " + PythonFunctions.describe(t), t); + } + } + + @Override + public boolean onBeforeCreate(final Record record) { + return dispatch(record); + } + + @Override + public boolean onBeforeUpdate(final Record record) { + return dispatch(record); + } + + @Override + public boolean onBeforeDelete(final Record record) { + return dispatch(record); + } + + @Override + public void onAfterCreate(final Record record) { + dispatch(record); + } + + @Override + public void onAfterUpdate(final Record record) { + dispatch(record); + } + + @Override + public void onAfterDelete(final Record record) { + dispatch(record); + } + } + + private PythonRecordEvents() { + } + + /** + * Registers {@code callback} for {@code event} (before_create, after_create, before_update, after_update, + * before_delete or after_delete) on {@code events}, the registry of a database or of one type. Returns the handle that + * {@link #unregister} takes. + */ + public static Object register(final RecordEvents events, final String event, final Callback callback) { + final Listener listener = new Listener(event, callback); + switch (event) { + case "before_create" -> events.registerListener((BeforeRecordCreateListener) listener); + case "before_update" -> events.registerListener((BeforeRecordUpdateListener) listener); + case "before_delete" -> events.registerListener((BeforeRecordDeleteListener) listener); + case "after_create" -> events.registerListener((AfterRecordCreateListener) listener); + case "after_update" -> events.registerListener((AfterRecordUpdateListener) listener); + case "after_delete" -> events.registerListener((AfterRecordDeleteListener) listener); + default -> throw new IllegalArgumentException("Unknown record event: " + event); + } + return listener; + } + + /** Removes the listener {@code handle} from {@code events}. */ + public static void unregister(final RecordEvents events, final String event, final Object handle) { + final Listener listener = (Listener) handle; + switch (event) { + case "before_create" -> events.unregisterListener((BeforeRecordCreateListener) listener); + case "before_update" -> events.unregisterListener((BeforeRecordUpdateListener) listener); + case "before_delete" -> events.unregisterListener((BeforeRecordDeleteListener) listener); + case "after_create" -> events.unregisterListener((AfterRecordCreateListener) listener); + case "after_update" -> events.unregisterListener((AfterRecordUpdateListener) listener); + case "after_delete" -> events.unregisterListener((AfterRecordDeleteListener) listener); + default -> throw new IllegalArgumentException("Unknown record event: " + event); + } + } +} diff --git a/tests/conftest.py b/tests/conftest.py index b04109d..9bf0321 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,6 +16,8 @@ if sys.platform == "win32": # test_sigint.py sends SIGINT to a child process, which Windows cannot deliver collect_ignore.append("test_sigint.py") + # test_python_function_interrupt.py sends SIGINT to its own process, which Windows cannot do either + collect_ignore.append("test_python_function_interrupt.py") # runs scripts/stage_jars.sh under bash with a fake docker; the bash a Windows # subprocess finds may be WSL's rather than Git Bash collect_ignore.append("test_stage_jars.py") diff --git a/tests/test_open_read_only.py b/tests/test_open_read_only.py index 51c4da0..c26b1e0 100644 --- a/tests/test_open_read_only.py +++ b/tests/test_open_read_only.py @@ -6,7 +6,6 @@ statement (see `test_sql_constraints_and_maintenance.py`). """ -import jpype import pytest import arcadedb_embedded as arcadedb @@ -46,12 +45,12 @@ def test_read_only_open_reads_and_refuses_writes(populated_path): def test_read_only_begin_is_not_refused_the_write_and_the_commit_are(populated_path): """begin() returns on a read-only database; the first write raises, and so does the commit - (the engine starts the transaction then). A write through Document.save() reaches - the engine's own exception. Reads need no transaction.""" + (the engine starts the transaction then). A write through Document.save() is an + ArcadeDBError like the rest of the API. Reads need no transaction.""" with arcadedb.open_database(populated_path, read_only=True) as db: db.begin() # not refused assert db.count_type("Note") == 2 - with pytest.raises(jpype.JException, match="(?i)read.?only"): + with pytest.raises(ArcadeDBError, match="(?i)read.?only"): db.new_document("Note").set("n", 3).save() with pytest.raises(ArcadeDBError, match="(?i)read.?only"): db.commit() diff --git a/tests/test_python_callbacks.py b/tests/test_python_callbacks.py new file mode 100644 index 0000000..53dd205 --- /dev/null +++ b/tests/test_python_callbacks.py @@ -0,0 +1,639 @@ +""" +Python functions called from SQL and openCypher (``Database.register_function``) and +Python callbacks on record events (``Database.on``). +""" + +import sys +import threading +from datetime import datetime +from decimal import Decimal + +import pytest + +import arcadedb_embedded as arcadedb +from arcadedb_embedded.callbacks import EVENTS + + +@pytest.fixture +def db(temp_db_path): + with arcadedb.create_database(temp_db_path) as database: + database.command("sql", "CREATE DOCUMENT TYPE Person") + database.command("sql", "CREATE DOCUMENT TYPE Pet") + database.command("sql", "CREATE VERTEX TYPE City") + yield database + + +def _insert_people(db, names): + with db.transaction(): + for name in names: + db.command("sql", "INSERT INTO Person SET name = ?", name) + + +def _names(db, type_name="Person"): + statement = f"SELECT name FROM {type_name} ORDER BY name" # nosec B608 - literal + rows = db.query("sql", statement).to_list() + return [row["name"] for row in rows] + + +# --------------------------------------------------------------------------- functions + + +def test_function_in_select_and_where(db): + _insert_people(db, ["Ann Lee", "Bob Ray"]) + db.register_function("slugify", lambda s: s.lower().replace(" ", "-")) + + rows = db.query("sql", "SELECT slugify(name) AS s FROM Person ORDER BY s").to_list() + assert [row["s"] for row in rows] == ["ann-lee", "bob-ray"] + + rows = db.query( + "sql", "SELECT name FROM Person WHERE slugify(name) = 'bob-ray'" + ).to_list() + assert [row["name"] for row in rows] == ["Bob Ray"] + + +def test_function_name_is_case_insensitive(db): + db.register_function("shout", lambda s: s.upper() + "!") + row = db.query("sql", "SELECT SHOUT('a') AS s").first() + assert row.get("s") == "A!" + + +def test_function_receives_a_record_as_a_wrapper(db): + _insert_people(db, ["Ann"]) + kinds = [] + + def name_of(record): + kinds.append(type(record).__name__) + return record.get("name") + + db.register_function("name_of", name_of) + row = db.query("sql", "SELECT name_of(@this) AS n FROM Person").first() + assert row.get("n") == "Ann" + assert kinds == ["Document"] + + +def test_function_receives_null_as_none_and_none_is_null(db): + _insert_people(db, ["Ann", None]) + seen = [] + + def describe(value): + seen.append(value) + return None if value is None else value.upper() + + db.register_function("describe", describe) + rows = db.query("sql", "SELECT describe(name) AS d FROM Person").to_list() + assert sorted(seen, key=lambda v: v or "") == [None, "Ann"] + assert sorted(row["d"] or "" for row in rows) == ["", "ANN"] + assert ( + db.query("sql", "SELECT FROM Person WHERE describe(name) IS NULL").to_list()[0][ + "name" + ] + is None + ) + + +def test_function_result_types(db): + db.register_function("an_int", lambda: 7) + db.register_function("a_float", lambda: 1.5) + db.register_function("a_bool", lambda: True) + db.register_function("a_text", lambda: "x") + db.register_function("a_decimal", lambda: Decimal("1.25")) + db.register_function("a_list", lambda: [1, 2, 3]) + db.register_function("a_dict", lambda: {"k": [1, "v"]}) + db.register_function("a_time", lambda: datetime(2026, 1, 2, 3, 4, 5)) + row = db.query( + "sql", + "SELECT an_int() AS i, a_float() AS f, a_bool() AS b, a_text() AS t, " + "a_decimal() AS d, a_list() AS l, a_dict() AS m, a_time() AS ts", + ).first() + assert row.get("i") == 7 + assert row.get("f") == 1.5 + assert row.get("b") is True + assert row.get("t") == "x" + assert row.get("d") == Decimal("1.25") + assert row.get("l") == [1, 2, 3] + assert row.get("m") == {"k": [1, "v"]} + assert row.get("ts") == datetime(2026, 1, 2, 3, 4, 5) + + +def test_function_arguments_arrive_as_python_values(db): + got = [] + db.register_function("capture", lambda *args: got.append(args) or 0) + db.query( + "sql", "SELECT capture(1, 2.5, 'a', true, [1, 2], {'k': 'v'}, null) AS c" + ).to_list() + assert got == [(1, 2.5, "a", True, [1, 2], {"k": "v"}, None)] + + +def test_function_exception_becomes_query_error(db): + _insert_people(db, ["Ann"]) + + def explode(value): + raise ValueError(f"cannot handle {value}") + + db.register_function("explode", explode) + with pytest.raises(arcadedb.ArcadeDBError) as error: + db.query("sql", "SELECT explode(name) FROM Person").to_list() + assert "explode" in str(error.value) + assert "ValueError: cannot handle Ann" in str(error.value) + + # the database, and the function table, work as before + assert db.query("sql", "SELECT 1 + 1 AS two").first().get("two") == 2 + + +def test_function_base_exception_does_not_reach_the_jvm(db): + def leave(): + raise SystemExit(3) + + db.register_function("leave", leave) + with pytest.raises(arcadedb.ArcadeDBError, match="SystemExit"): + db.query("sql", "SELECT leave() AS x").to_list() + assert db.query("sql", "SELECT 1 + 1 AS two").first().get("two") == 2 + + +def test_keyboard_interrupt_in_a_function_reaches_the_caller_as_one(db): + """Ctrl-C always arrives as KeyboardInterrupt, chained from the ArcadeDBError that names the + function; the statement is aborted and the database works as before.""" + _insert_people(db, ["Ann", "Bob"]) + + def interrupt(value): + raise KeyboardInterrupt("stop") + + db.register_function("interrupt", interrupt) + for read in ( + lambda: db.query("sql", "SELECT interrupt(name) AS x FROM Person").to_list(), + lambda: list(db.query("sql", "SELECT interrupt(name) AS x FROM Person")), + lambda: db.command("sql", "SELECT interrupt(name) AS x FROM Person").to_list(), + ): + with pytest.raises(KeyboardInterrupt) as error: + read() + assert isinstance(error.value.__cause__, arcadedb.ArcadeDBError) + assert "interrupt" in str(error.value.__cause__) + assert _names(db) == ["Ann", "Bob"] + + +def test_a_stale_interrupt_does_not_turn_a_later_error_into_one(db): + def interrupt(value): + raise KeyboardInterrupt("stop") + + db.register_function("interrupt", interrupt) + with pytest.raises(KeyboardInterrupt): + db.query("sql", "SELECT interrupt(1) AS x").to_list() + with pytest.raises(arcadedb.ArcadeDBError): + db.query("sql", "SELEC nonsense") + + +def test_unconvertible_return_values_are_readable_errors(db): + import numpy as np + + for label, value in ( + ("TypeError", np.array([1, 2])), + ("OverflowError", 2**63), + ("TypeError", object()), + ): + db.register_function("bad", lambda x, value=value: value, replace=True) + with pytest.raises(arcadedb.ArcadeDBError) as error: + db.query("sql", "SELECT bad(1) AS x").to_list() + assert "bad" in str(error.value) + assert label in str(error.value) + assert "PyExceptionProxy" not in str(error.value) + db.register_function("bad", lambda x: 2**62, replace=True) + assert db.query("sql", "SELECT bad(1) AS x").first().get("x") == 2**62 + + +@pytest.mark.xfail( + not getattr(sys, "_is_gil_enabled", lambda: True)(), + reason=( + "free-threaded CPython only: the Python heap grows about 520 to 610 bytes per " + "register, on, remove cycle, cause not attributed (JPype proxy cleanup suspected, " + "jpype-project/jpype#1506); see 'Free-threaded Python' in the bindings documentation" + ), + strict=False, +) +def test_registering_again_and_again_does_not_grow_the_python_heap(db): + """The JPype proxy class is defined once, not per registration (it leaked about 6 KB each).""" + import gc + import time + import tracemalloc + + import jpype + + def settle(): + # JPype frees a proxy's Python object once the JVM has collected the Java side + jpype.JClass("java.lang.System").gc() + time.sleep(0.3) + gc.collect() + + def cycle(n): + for _ in range(n): + db.register_function("churn", lambda x: x, replace=True) + handle = db.on("after_create", lambda record: None, "Person") + handle.remove() + db.unregister_function("churn") + + cycle(300) # warm-up: caches, the first proxy classes + settle() + tracemalloc.start() + try: + before, _ = tracemalloc.get_traced_memory() + cycle(2000) + settle() + after, _ = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert ( + after - before < 1_000_000 + ) # it was about 24,000,000 when each call defined a class + + +def test_function_argument_count_is_inferred(db): + def two_or_three(a, b, c=0): + return a + b + c + + db.register_function("sum_args", two_or_three) + assert db.query("sql", "SELECT sum_args(1, 2) AS s").first().get("s") == 3 + assert db.query("sql", "SELECT sum_args(1, 2, 3) AS s").first().get("s") == 6 + for bad in ("SELECT sum_args(1)", "SELECT sum_args(1, 2, 3, 4)"): + with pytest.raises(arcadedb.ArcadeDBError, match="sum_args"): + db.query("sql", bad).to_list() + + db.register_function("count_args", lambda *args: len(args)) + assert ( + db.query("sql", "SELECT count_args(1, 2, 3, 4, 5) AS n").first().get("n") == 5 + ) + + db.register_function("exactly_one", lambda x, y=0: x, max_args=1) + with pytest.raises(arcadedb.ArcadeDBError, match="exactly_one"): + db.query("sql", "SELECT exactly_one(1, 2)").to_list() + + +def test_function_registration_errors(db): + with pytest.raises(ValueError): + db.register_function("not a name", lambda: 1) + with pytest.raises(ValueError): + db.register_function("1abc", lambda: 1) + with pytest.raises(TypeError): + db.register_function("nope", "not callable") + with pytest.raises(ValueError): + db.register_function("bad_counts", lambda *a: 1, min_args=3, max_args=1) + with pytest.raises(arcadedb.ArcadeDBError, match="keyword-only"): + db.register_function("kw_only", lambda *, key: key) + for engine_name in ("abs", "count", "toUpper", "id"): + with pytest.raises(arcadedb.ArcadeDBError, match="already a function"): + db.register_function(engine_name, lambda *a: 1) + # the engine's own functions still work after the refusals + assert db.query("sql", "SELECT abs(-3) AS a").first().get("a") == 3 + + +def test_function_replace(db): + db.register_function("which", lambda: "first") + with pytest.raises(arcadedb.ArcadeDBError, match="replace=True"): + db.register_function("which", lambda: "second") + assert db.query("sql", "SELECT which() AS w").first().get("w") == "first" + + db.register_function("which", lambda: "second", replace=True) + assert db.query("sql", "SELECT which() AS w").first().get("w") == "second" + + +def test_unregister_function(db): + db.register_function("temp_fn", lambda: 1) + statement = "SELECT temp_fn() AS x" + assert db.query("sql", statement).first().get("x") == 1 + + db.unregister_function("temp_fn") + # the same statement text, which may have a cached plan, fails now + with pytest.raises(arcadedb.ArcadeDBError, match="temp_fn"): + db.query("sql", statement).to_list() + with pytest.raises(arcadedb.ArcadeDBError, match="No Python function"): + db.unregister_function("temp_fn") + + # the name can be registered again + db.register_function("temp_fn", lambda: 2) + assert db.query("sql", statement).first().get("x") == 2 + + +def test_close_unregisters_the_functions_of_a_database(temp_db_path): + with arcadedb.create_database(temp_db_path) as first: + first.register_function("scoped_fn", lambda: 1) + assert first.query("sql", "SELECT scoped_fn() AS x").first().get("x") == 1 + with arcadedb.open_database(temp_db_path) as second: + with pytest.raises(arcadedb.ArcadeDBError, match="scoped_fn"): + second.query("sql", "SELECT scoped_fn() AS x").to_list() + # and the name is free again + second.register_function("scoped_fn", lambda: 2) + assert second.query("sql", "SELECT scoped_fn() AS x").first().get("x") == 2 + + +def test_function_in_opencypher_needs_a_namespaced_name(db): + with db.transaction(): + db.command("sql", "CREATE VERTEX City SET name = 'New York'") + db.register_function("py.slugify", lambda s: s.lower().replace(" ", "-")) + + rows = db.query( + "opencypher", "MATCH (c:City) RETURN py.slugify(c.name) AS s" + ).to_list() + assert rows == [{"s": "new-york"}] + # the same function from SQL + row = db.query("sql", "SELECT py.slugify(name) AS s FROM City").first() + assert row.get("s") == "new-york" + + # a bare name is rejected by the openCypher validator before it reaches the registry + db.register_function("bare_slugify", lambda s: s) + with pytest.raises(arcadedb.ArcadeDBError, match="Unknown function"): + db.query("opencypher", "MATCH (c:City) RETURN bare_slugify(c.name)").to_list() + + +def test_function_failure_in_opencypher_is_a_query_error(db): + with db.transaction(): + db.command("sql", "CREATE VERTEX City SET name = 'Oslo'") + + def explode(_): + raise RuntimeError("cypher boom") + + db.register_function("py.explode", explode) + with pytest.raises(arcadedb.ArcadeDBError, match="RuntimeError: cypher boom"): + db.query("opencypher", "MATCH (c:City) RETURN py.explode(c.name)").to_list() + + +def test_function_called_from_an_engine_thread(db): + thread_names = [] + + def mark(value): + thread_names.append(threading.current_thread().name) + return value + + db.register_function("mark", mark) + with db.transaction(): + db.command("sql", "INSERT INTO Person SET name = 'x'") + executor = db.async_executor() + executor.command("sql", "INSERT INTO Pet SET name = mark('from async')") + executor.wait_completion() + assert _names(db, "Pet") == ["from async"] + assert thread_names and all(isinstance(name, str) for name in thread_names) + + +def test_function_called_concurrently_from_python_threads(db): + _insert_people(db, [f"n{i}" for i in range(50)]) + db.register_function("twice", lambda s: s + s) + results, failures = [], [] + + def run(): + try: + rows = db.query("sql", "SELECT twice(name) AS t FROM Person").to_list() + results.append(len(rows)) + except Exception as e: # pragma: no cover - reported by the assertion + failures.append(e) + + threads = [threading.Thread(target=run) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert not failures + assert results == [50, 50, 50, 50] + + +# ------------------------------------------------------------------------------ events + + +def test_each_event_fires_once_per_change_with_the_record(db): + log = [] + for event in EVENTS: + db.on( + event, + lambda record, event=event: log.append( + (event, record.get_type_name(), record.get("name")) + ), + "Person", + ) + + with db.transaction(): + db.command("sql", "INSERT INTO Person SET name = 'Ann'") + assert log == [ + ("before_create", "Person", "Ann"), + ("after_create", "Person", "Ann"), + ] + + del log[:] + with db.transaction(): + db.command("sql", "UPDATE Person SET name = 'Anna' WHERE name = 'Ann'") + assert [entry[0] for entry in log] == ["before_update", "after_update"] + assert log[0][2] == "Anna" + + del log[:] + with db.transaction(): + db.command("sql", "DELETE FROM Person WHERE name = 'Anna'") + assert log == [ + ("before_delete", "Person", "Anna"), + ("after_delete", "Person", "Anna"), + ] + + +def test_events_fire_for_the_record_api_too(db): + created, updated, deleted = [], [], [] + db.on("after_create", created.append, "Person") + db.on("after_update", updated.append, "Person") + db.on("after_delete", deleted.append, "Person") + + with db.transaction(): + document = db.new_document("Person") + document.set("name", "Zed") + document.save() + assert [r.get("name") for r in created] == ["Zed"] + assert isinstance(created[0], arcadedb.Document) + rid = created[0].get_rid() + + with db.transaction(): + record = db.lookup_by_rid(rid) + record.modify().set("name", "Zoe").save() + assert [r.get("name") for r in updated] == ["Zoe"] + + with db.transaction(): + db.lookup_by_rid(rid).delete() + assert [r.get_rid() for r in deleted] == [rid] + + +def test_events_fire_for_opencypher_writes(db): + created = [] + db.on("after_create", lambda r: created.append(r.get("name")), "City") + with db.transaction(): + db.command("opencypher", "CREATE (:City {name: 'Oslo'})") + assert created == ["Oslo"] + + +def test_event_passes_vertex_wrappers(db): + seen = [] + db.on("after_create", seen.append, "City") + with db.transaction(): + vertex = db.new_vertex("City") + vertex.set("name", "Oslo") + vertex.save() + assert len(seen) == 1 and isinstance(seen[0], arcadedb.Vertex) + + +def test_type_filter_limits_the_events(db): + on_person, on_everything = [], [] + db.on("after_create", lambda r: on_person.append(r.get_type_name()), "Person") + db.on("after_create", lambda r: on_everything.append(r.get_type_name())) + + with db.transaction(): + db.command("sql", "INSERT INTO Person SET name = 'a'") + db.command("sql", "INSERT INTO Pet SET name = 'b'") + db.command("sql", "CREATE VERTEX City SET name = 'c'") + assert on_person == ["Person"] + assert sorted(on_everything) == ["City", "Person", "Pet"] + + +def test_type_filter_survives_schema_changes(db): + seen = [] + db.on("after_create", lambda r: seen.append(r.get("name")), "Person") + db.command("sql", "CREATE PROPERTY Person.name STRING") + db.command("sql", "CREATE INDEX ON Person (name) NOTUNIQUE") + db.command("sql", "CREATE DOCUMENT TYPE Unrelated") + _insert_people(db, ["after schema change"]) + assert seen == ["after schema change"] + + +def test_before_create_veto(db): + db.on("before_create", lambda r: r.get("name") != "blocked", "Person") + with db.transaction(): + db.command("sql", "INSERT INTO Person SET name = 'ok'") + db.command("sql", "INSERT INTO Person SET name = 'blocked'") + document = db.new_document("Person") + document.set("name", "blocked") + document.save() + assert _names(db) == ["ok"] + + +def test_before_update_veto(db): + _insert_people(db, ["keep", "change"]) + db.on("before_update", lambda r: r.get("name") != "frozen", "Person") + with db.transaction(): + db.command("sql", "UPDATE Person SET name = 'frozen' WHERE name = 'keep'") + db.command("sql", "UPDATE Person SET name = 'changed' WHERE name = 'change'") + assert _names(db) == ["changed", "keep"] + + +def test_before_delete_veto(db): + _insert_people(db, ["protected", "removable"]) + db.on("before_delete", lambda r: r.get("name") != "protected", "Person") + with db.transaction(): + db.command("sql", "DELETE FROM Person") + assert _names(db) == ["protected"] + + +def test_none_and_true_allow_and_after_result_is_ignored(db): + db.on("before_create", lambda r: None, "Person") + db.on("before_create", lambda r: True, "Person") + db.on("after_create", lambda r: False, "Person") + _insert_people(db, ["a"]) + assert _names(db) == ["a"] + + +def test_remove_stops_the_callback(db): + seen = [] + handle = db.on("after_create", lambda r: seen.append(r.get("name")), "Person") + assert handle.active and handle.event == "after_create" + assert handle.type_name == "Person" + _insert_people(db, ["one"]) + + handle.remove() + assert not handle.active + handle.remove() # a second call does nothing + _insert_people(db, ["two"]) + assert seen == ["one"] + assert "removed" in repr(handle) + + +def test_remove_database_wide_and_context_manager(db): + seen = [] + with db.on("after_create", lambda r: seen.append(r.get("name"))) as handle: + _insert_people(db, ["in"]) + assert not handle.active + _insert_people(db, ["out"]) + assert seen == ["in"] + + +def test_two_callbacks_on_one_event_both_fire(db): + first, second = [], [] + db.on("after_create", first.append, "Person") + db.on("after_create", second.append, "Person") + _insert_people(db, ["x"]) + assert len(first) == 1 and len(second) == 1 + + +def test_before_callback_exception_fails_the_operation(db): + def explode(record): + raise RuntimeError("no creating today") + + db.on("before_create", explode, "Person") + with pytest.raises(arcadedb.ArcadeDBError) as error: + _insert_people(db, ["x"]) + assert "before_create" in str(error.value) + assert "RuntimeError: no creating today" in str(error.value) + assert db.count_type("Person") == 0 + + +def test_after_callback_exception_rolls_the_transaction_back(db): + def explode(record): + raise RuntimeError("after boom") + + handle = db.on("after_create", explode, "Person") + with pytest.raises(arcadedb.ArcadeDBError, match="after_create.*after boom"): + _insert_people(db, ["x", "y"]) + assert db.count_type("Person") == 0 + + handle.remove() + _insert_people(db, ["z"]) + assert _names(db) == ["z"] + + +def test_save_and_delete_raise_arcadedb_error_when_a_callback_fails(db): + """Document.save() and delete() wrap the engine's Java exceptions like the rest of the API.""" + + def explode(record): + raise RuntimeError("no today") + + handle = db.on("before_create", explode, "Person") + with pytest.raises(arcadedb.ArcadeDBError, match="before_create.*no today"): + with db.transaction(): + db.new_document("Person").set("name", "x").save() + handle.remove() + assert _names(db) == [] + + with db.transaction(): + db.new_document("Person").set("name", "y").save() + handle = db.on("before_delete", explode, "Person") + with pytest.raises(arcadedb.ArcadeDBError, match="before_delete.*no today"): + with db.transaction(): + db.query("sql", "SELECT FROM Person").first().get_element().delete() + handle.remove() + assert _names(db) == ["y"] + + # a veto (False) is not an error: the record is just not written + veto = db.on("before_create", lambda record: False, "Person") + with db.transaction(): + db.new_document("Person").set("name", "z").save() + veto.remove() + assert _names(db) == ["y"] + + # without a transaction the engine refuses, as an ArcadeDBError + with pytest.raises(arcadedb.ArcadeDBError): + db.new_document("Person").set("name", "w").save() + + +def test_event_registration_errors(db): + with pytest.raises(ValueError, match="Invalid event"): + db.on("on_create", lambda r: None) + with pytest.raises(TypeError): + db.on("after_create", "not callable") + with pytest.raises(arcadedb.ArcadeDBError, match="does not exist"): + db.on("after_create", lambda r: None, "NoSuchType") + + +def test_calls_after_close_are_refused(temp_db_path): + database = arcadedb.create_database(temp_db_path) + database.close() + with pytest.raises(arcadedb.ArcadeDBError, match="closed"): + database.register_function("late", lambda: 1) + with pytest.raises(arcadedb.ArcadeDBError, match="closed"): + database.on("after_create", lambda r: None) diff --git a/tests/test_python_function_interrupt.py b/tests/test_python_function_interrupt.py new file mode 100644 index 0000000..36d624c --- /dev/null +++ b/tests/test_python_function_interrupt.py @@ -0,0 +1,84 @@ +"""A real Ctrl-C during a statement that calls a Python function always arrives as KeyboardInterrupt. + +The signal can land in the function, in the engine between two calls, or on entering the callback +(where the exception leaves it as JPype's PyExceptionProxy and the engine wraps it). Each of these +used to come out differently (KeyboardInterrupt, or ArcadeDBError). When the signal also lands while +JPype converts the engine's exception, JPype fails with ``RuntimeError: Fatal error occurred`` +(jpype-project/jpype#1496), which was an ArcadeDBError in about 5% of the runs; it is a +KeyboardInterrupt now. Not run on Windows, where a +signal cannot be sent to the own process like this (see test_sigint.py). +""" + +import os +import signal +import threading +import time + +import numpy as np +import pytest + +import arcadedb_embedded as arcadedb + + +@pytest.fixture +def db(temp_db_path): + with arcadedb.create_database(temp_db_path) as database: + database.command("sql", "CREATE DOCUMENT TYPE T") + database.insert_columns("T", {"n": np.arange(300_000, dtype=np.int64)}) + database.register_function("plus_one", lambda x: x + 1) + yield database + + +def test_sigint_during_a_query_that_calls_python_is_a_keyboard_interrupt(db): + interrupted = 0 + for delay in (0.05, 0.09, 0.13, 0.17, 0.21, 0.25, 0.29, 0.33): + timer = threading.Timer(delay, os.kill, (os.getpid(), signal.SIGINT)) + timer.start() + try: + db.query("sql", "SELECT sum(plus_one(n)) AS s FROM T").first() + outcome = "completed" + except KeyboardInterrupt: + outcome = "interrupted" + interrupted += 1 + finally: + timer.cancel() + timer.join() + if outcome == "completed": + try: # a signal sent just after the end is delivered here, still a KeyboardInterrupt + time.sleep(0.05) + except KeyboardInterrupt: + interrupted += 1 + # the database works as before + assert db.count_type("T") == 300_000 + assert interrupted >= 1 + + +class _JpypeLostTheRace: + """Stands in for the Java database when JPype loses its race with Ctrl-C (jpype#1496). + + The engine handles a KeyboardInterrupt that left the Python function as a Java exception; + JPype's conversion of it to Python fails with ``InterruptedException: sleep interrupted`` + and the call raises this plain RuntimeError, with nothing left for the bindings to read. + """ + + def query(self, *args): + raise RuntimeError("Fatal error occurred") + + command = query + + +@pytest.mark.parametrize("method", ["query", "command"]) +def test_jpypes_fatal_error_during_a_statement_is_a_keyboard_interrupt( + db, method, monkeypatch +): + monkeypatch.setattr(db, "_java_db", _JpypeLostTheRace()) + with pytest.raises(KeyboardInterrupt) as raised: + getattr(db, method)("sql", "SELECT sum(plus_one(n)) AS s FROM T") + # the text of JPype's error stays reachable + assert "Fatal error occurred" in str(raised.value.__cause__) + + +def test_a_statement_that_fails_otherwise_is_still_an_arcadedb_error(db): + with pytest.raises(arcadedb.ArcadeDBError) as raised: + db.query("sql", "SELECT FROM NoSuchType") + assert not isinstance(raised.value, KeyboardInterrupt)