Visitar URL original
arcadedb.aio.AsyncDatabase: await query, command, run_in_transaction … · ArcadeData/arcadedb-python@65e7c8f · GitHub
Skip to content

Commit 65e7c8f

Browse files
tae898claude
andauthored
arcadedb.aio.AsyncDatabase: await query, command, run_in_transaction and find_nearest from async def code (refs #306) (#379)
* arcadedb.aio.AsyncDatabase: await query, command, run_in_transaction and find_nearest from async def code (refs #306) An asyncio facade over the blocking calls of an open Database, on a bounded thread pool of its own (max_workers, default 8). Decided on #306: a transaction belongs to one thread and two awaits of one task can run on two workers, so there is no async with transaction; run_in_transaction(fn) runs the plain function fn, begin to commit and every retry, on one worker. Cancellation is not exposed: cancelling the awaiting task abandons the result, a call that started finishes, a call still waiting for a worker does not run, and close() waits for the calls that were submitted (it does not close the Database). The docstring and the docs page say how AsyncDatabase differs from the engine's AsyncExecutor. tests/test_aio.py pins, on both wheels: rows and errors reach the awaiting task, run_in_transaction commits on one worker and rolls back on an exception, 200 tasks write and read at once without a lost update, the pool runs max_workers calls at once and never more, close() waits and refuses new calls, a cancelled await abandons the result. tests/test_docs_asyncio_page.py runs the blocks of the new page how-to/asyncio.adoc. Items 3 and 4 of #306 still need measurements. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> * Read-your-writes tests also through AsyncDatabase; the concurrency and vectors pages point to the facade (refs #306) The two read-your-writes tests of #373 (the writing task, and other tasks that search after the commit) run again through arcadedb.aio.AsyncDatabase: an insert is run_in_transaction and a search is find_nearest, on both wheels. The asyncio section of the concurrency page and the vectors page name the facade next to asyncio.to_thread. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> * Test helper: a nosec note for the fixed type name (Bandit B608) The count helper builds its SQL from a fixed type name of the tests, backtick-quoted, with no values. CI's Bandit run flags it at low confidence; the repo's convention is a nosec with the reason. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
1 parent 085cfff commit 65e7c8f

9 files changed

Lines changed: 838 additions & 3 deletions

File tree

‎docs/modules/ROOT/nav.adoc‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
** xref:how-to/transactions.adoc[Transactions]
88
** xref:how-to/values-and-records.adoc[Values and records]
99
** xref:how-to/concurrency.adoc[Concurrency, processes and crashes]
10+
** xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]
1011
** xref:how-to/durability.adoc[Durability settings]
1112
** xref:how-to/async-executor.adoc[The async executor]
1213
** xref:how-to/logs.adoc[Where the logs go]

‎docs/modules/ROOT/pages/how-to/async-executor.adoc‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33

44
// Sources: src/arcadedb/async_executor.py (the JVM build), src/arcadedb/_native/extras.py (NativeAsyncExecutor, the native build), core.py (async_executor), and ArcadeDB 26.11.1 DatabaseAsyncExecutorImpl.java. Pinned by tests/test_async_executor.py, tests/test_async_transaction.py and tests/test_async_executor_behavior.py.
55

6-
You need this page if you want to hand statements to the engine and carry on, and learn about the result in a callback. The short version:
6+
You need this page if you want to hand statements to the engine and carry on, and learn about the result in a callback. For `async def` code that awaits each call, use `AsyncDatabase` instead: see xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]. The short version:
77

88
* `executor = db.async_executor()` returns a wrapper of the one executor of the database. Every call returns another wrapper of the same executor, with the same settings.
99
* `executor.command(...)`, `executor.query(...)` and `executor.transaction(...)` queue work and return at once. `executor.wait_completion()` blocks until everything queued before it is stored.
Lines changed: 221 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,221 @@
1+
= asyncio: AsyncDatabase
2+
:description: Call a Database from async def code with arcadedb.aio.AsyncDatabase: awaited query, command, run_in_transaction and find_nearest on a bounded thread pool, what a transaction, a cancel and close() mean there, and how it differs from the async executor.
3+
4+
// Sources: src/arcadedb/aio.py. Pinned by tests/test_aio.py, tests/test_docs_asyncio_page.py and tests/test_vector_read_your_writes.py.
5+
6+
You need this page if your code is written with `async def`, for example an agent loop that stores a few vectors per turn and searches right after, and a database call must not block the event loop. The short version:
7+
8+
* `adb = AsyncDatabase(db, max_workers=8)` wraps an open `Database`. It owns a pool of `max_workers` threads, and it is a thin layer: the calls and the errors are those of `db`.
9+
* `await adb.query(...)`, `await adb.command(...)` and `await adb.find_nearest(index, ...)` run the blocking call on a worker and return its result. `await adb.run(fn, *args)` does the same for any function.
10+
* A transaction belongs to one thread, and two awaits of one task can run on two workers. So there is no `async with adb.transaction()`: `await adb.run_in_transaction(fn)` runs the plain function `fn`, from begin to commit and through every retry, on one worker.
11+
* Cancelling the task that awaits a call abandons the result. A call that has started on a worker still finishes. A call that still waits for a worker does not run.
12+
* `await adb.close()` refuses new calls and waits for the ones that were submitted. It does not close the `Database`.
13+
14+
The statements are SQL or openCypher, as on xref:how-to/databases-and-queries.adoc[Databases and queries]. This page covers what the facade adds. What the engine does with several threads is on xref:how-to/concurrency.adoc[Concurrency, processes and crashes].
15+
16+
The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way (`tests/test_docs_asyncio_page.py`) on the JVM build and on the native build. Each block that prints something is followed by its output. Every block assumes that `db` is an open database.
17+
18+
== Open it, write and read
19+
20+
`AsyncDatabase` is an async context manager, so `close()` runs when the block ends. A write goes in `run_in_transaction`, because a transaction cannot span awaits. A read needs no transaction. `query` returns a list of `dict`, the rows as `ResultSet.to_list()` gives them, read on the worker.
21+
22+
[source,python]
23+
----
24+
import asyncio
25+
26+
from arcadedb.aio import AsyncDatabase
27+
28+
db.command("sql", "CREATE DOCUMENT TYPE Note")
29+
30+
31+
def add_hello():
32+
db.command("sql", "INSERT INTO Note SET text = ?", "hello")
33+
34+
35+
async def main():
36+
async with AsyncDatabase(db, max_workers=4) as adb:
37+
await adb.run_in_transaction(add_hello)
38+
rows = await adb.query("sql", "SELECT text FROM Note")
39+
print([row["text"] for row in rows])
40+
41+
42+
asyncio.run(main())
43+
----
44+
45+
[source,text]
46+
----
47+
['hello']
48+
----
49+
50+
Parameters bind as in `db.query`: positional `?` from the extra arguments, or by name from one `dict`. `command` returns a list of `dict` as well, or `None` when the engine returns no result set. An `INSERT`, `UPDATE` or `DELETE` through `command` needs a transaction, so send it through `run_in_transaction`, or switch on `db.set_auto_transaction(True)`, which commits each statement on its own (see xref:how-to/transactions.adoc#auto-transaction[Transactions]). Do not send `BEGIN` or `COMMIT` through `command`: the next call can run on another worker, where that transaction is not open.
51+
52+
== Many tasks at once
53+
54+
Start as many tasks as you like. The pool runs at most `max_workers` calls at a time, and the others wait for a worker. Writers that change the same record conflict, and `run_in_transaction` runs the function again after such a conflict: pass `retries` for a contended record, as on xref:how-to/concurrency.adoc#several-writers[Several writers].
55+
56+
[source,python]
57+
----
58+
async def remember(adb, n):
59+
def write():
60+
db.command("sql", "INSERT INTO Note SET text = ?", f"note {n}")
61+
62+
await adb.run_in_transaction(write, retries=60)
63+
64+
65+
async def main():
66+
async with AsyncDatabase(db) as adb:
67+
await asyncio.gather(*(remember(adb, n) for n in range(100)))
68+
rows = await adb.query("sql", "SELECT count(*) AS n FROM Note")
69+
print(rows[0]["n"])
70+
71+
72+
asyncio.run(main())
73+
----
74+
75+
[source,text]
76+
----
77+
101
78+
----
79+
80+
The default of 8 workers is a starting point, not a measured optimum. Measure it on your own workload before you change it.
81+
82+
== A transaction that fails
83+
84+
`fn` is a plain function with no arguments. It makes blocking calls on the `Database`, and it contains no `await`. An exception of `fn` rolls the transaction back and is raised in the awaiting task. The next block stores a note inside the function and then fails, and the note is gone.
85+
86+
[source,python]
87+
----
88+
def fail():
89+
db.command("sql", "INSERT INTO Note SET text = 'never stored'")
90+
raise ValueError("not this one")
91+
92+
93+
async def main():
94+
async with AsyncDatabase(db) as adb:
95+
try:
96+
await adb.run_in_transaction(fail)
97+
except ValueError as error:
98+
print("raised:", error)
99+
rows = await adb.query("sql", "SELECT count(*) AS n FROM Note")
100+
print(rows[0]["n"])
101+
102+
103+
asyncio.run(main())
104+
----
105+
106+
[source,text]
107+
----
108+
raised: not this one
109+
101
110+
----
111+
112+
== Any other call
113+
114+
`run(fn, *args, **kwargs)` runs any blocking function on a worker and returns its result. Use it for what the facade does not wrap, and for a result shape other than a list of `dict`. The context variables of the calling task are visible to `fn`, as with `asyncio.to_thread`.
115+
116+
[source,python]
117+
----
118+
def first_text():
119+
return db.query("sql", "SELECT text FROM Note ORDER BY text LIMIT 1").first().get("text")
120+
121+
122+
async def main():
123+
async with AsyncDatabase(db) as adb:
124+
print(await adb.run(first_text))
125+
126+
127+
asyncio.run(main())
128+
----
129+
130+
[source,text]
131+
----
132+
hello
133+
----
134+
135+
== Vector search
136+
137+
`find_nearest(index, query_vector, k=10, ...)` takes the `VectorIndex` that `db.create_vector_index(...)` or `db.schema.get_vector_index(...)` returns, and gives what `index.find_nearest` gives: a list of `(record, distance)`, nearest first. A search finds every vector whose transaction committed before the search started, so an agent can insert in one call and search in the next, from the same task or from another one: see xref:how-to/vectors.adoc#agents-that-write-as-they-go[Agents that write as they go].
138+
139+
[source,python]
140+
----
141+
db.command("sql", "CREATE VERTEX TYPE Memory")
142+
db.command("sql", "CREATE PROPERTY Memory.embedding ARRAY_OF_FLOATS")
143+
with db.transaction():
144+
for name, vector in [
145+
("red", [1.0, 0.0, 0.0]),
146+
("green", [0.0, 1.0, 0.0]),
147+
("blue", [0.0, 0.0, 1.0]),
148+
]:
149+
db.command(
150+
"sql", "INSERT INTO Memory SET name = ?, embedding = ?", name, vector
151+
)
152+
index = db.create_vector_index("Memory", "embedding", dimensions=3)
153+
154+
155+
def add_orange():
156+
db.command(
157+
"sql",
158+
"INSERT INTO Memory SET name = ?, embedding = ?",
159+
"orange",
160+
[0.9, 0.1, 0.0],
161+
)
162+
163+
164+
async def main():
165+
async with AsyncDatabase(db) as adb:
166+
await adb.run_in_transaction(add_orange)
167+
hits = await adb.find_nearest(index, [0.9, 0.1, 0.0], k=1)
168+
print(hits[0][0].get("name"))
169+
170+
171+
asyncio.run(main())
172+
----
173+
174+
[source,text]
175+
----
176+
orange
177+
----
178+
179+
== What runs where
180+
181+
Every call runs on a thread of the pool of the facade, never on the event loop, and the pool is its own: the facade does not use the default executor of the loop. The engine call runs without the Python GIL on both builds (JPype releases it for a Java call, `ctypes` for a call into the native library), so the loop keeps running while the engine works. Building the Python rows from the answer, which `query` and `command` do on the worker, holds the GIL. How late the loop wakes up during such calls was measured on xref:how-to/concurrency.adoc#asyncio[Concurrency, processes and crashes] for `asyncio.to_thread`, which is the same mechanism.
182+
183+
One `Database` object shared by the workers is supported on both builds: see xref:how-to/concurrency.adoc#threads[Threads]. A worker keeps nothing between two calls, as long as every transaction is opened and ended inside one `run_in_transaction`.
184+
185+
== Cancel and close
186+
187+
Cancellation is not part of the facade. Cancelling the task that awaits a call, with `task.cancel()` or a timeout, stops the wait and drops the result. What happens to the call depends on where it is:
188+
189+
* A call that still waits for a worker does not run.
190+
* A call that already runs on a worker is not interrupted, and finishes. A transaction function still commits or rolls back on its own worker, so no transaction is left half open.
191+
192+
`await adb.close()` sets the facade to closed, so a new call raises `RuntimeError`, and then waits for the calls that were submitted, the queued ones included. The wait does not block the event loop. A second `close()` waits for the same calls. The `Database` stays open: close it yourself.
193+
194+
== AsyncDatabase or the async executor
195+
196+
Both names say "async", and they are different tools.
197+
198+
[cols="2,3,3"]
199+
|===
200+
| | `AsyncDatabase` (`arcadedb.aio`) | `AsyncExecutor` (`db.async_executor()`)
201+
202+
| Made by
203+
| `AsyncDatabase(db, max_workers=8)`
204+
| `db.async_executor()`, the one executor of the database
205+
206+
| Runs on
207+
| Python threads of its own pool, through the blocking API
208+
| The threads of the engine
209+
210+
| A call
211+
| Awaited: returns its result, or raises in the awaiting task
212+
| Queued: returns at once, and the result arrives in a callback
213+
214+
| Transactions
215+
| `run_in_transaction(fn)`: one worker from begin to commit, with retries
216+
| Each worker commits after `commit_every` operations and at `wait_completion()`
217+
218+
| Use it for
219+
| `async def` code: reads, vector search and small transactions
220+
| Work that the engine runs in the background while your code carries on: see xref:how-to/async-executor.adoc[The async executor]
221+
|===

‎docs/modules/ROOT/pages/how-to/concurrency.adoc‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,8 @@ run_repeatable_read(db, lambda: add_one(3))
195195

196196
Call the bindings from asyncio through `asyncio.to_thread` or `loop.run_in_executor`. A call made directly on the event loop blocks the loop for as long as the call takes.
197197

198+
`arcadedb.aio.AsyncDatabase` does this wrapping for `query`, `command`, `run_in_transaction` and `find_nearest`, on a bounded thread pool of its own: see xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]. The probe below calls the bindings through `asyncio.to_thread` and `run_in_executor` directly.
199+
198200
The probe: 100 tasks, each making 20 operations (10 on Python 3.14t), so 2,000 operations (1,000 on Python 3.14t) per call style. One operation is `db.run_in_transaction(fn, retries=60)` with an atomic counter update and one unique ledger insert, then a count. A heartbeat task sleeps 10 ms in a loop and records how late it wakes up. Each cell below is that lateness in milliseconds, as the 99th percentile and the maximum. In the short phases, where the 99th percentile equals the maximum, there were too few heartbeats for them to differ. The two `async_executor()` rows use the bindings' `AsyncExecutor` with `set_parallel_level(1)` and `set_parallel_level(4)`: each task inserts one row with `command(..., callback=..., error_callback=...)`, and the callbacks hand the result to the loop with `loop.call_soon_threadsafe`.
199201

200202
[cols="4,2,2,2"]

‎docs/modules/ROOT/pages/how-to/vectors.adoc‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -508,7 +508,7 @@ print(notes.get_size())
508508
An agent that writes one record per turn and searches right after needs no setting: the rebuild runs in the background and the searches keep running while it does.
509509

510510
// Pinned by tests/test_vector_read_your_writes.py, on both packages: 200 inserts, each followed by a search, from the writing task and from other tasks.
511-
A search finds every vector whose transaction committed before the search started. Insert in one call and search in the next, from the same task or from another one, and the new vector is in the result, also while the graph rebuilds in the background. From asyncio, run each call with `asyncio.to_thread` as xref:how-to/concurrency.adoc#asyncio[Concurrency] describes: the insert and the search of one task can run on different threads and the search still finds the vector. This holds for committed writes only. It says nothing about a search made while the writing transaction is still open.
511+
A search finds every vector whose transaction committed before the search started. Insert in one call and search in the next, from the same task or from another one, and the new vector is in the result, also while the graph rebuilds in the background. From asyncio, use `AsyncDatabase` (xref:how-to/asyncio.adoc[asyncio: AsyncDatabase]) or run each call with `asyncio.to_thread` as xref:how-to/concurrency.adoc#asyncio[Concurrency] describes: the insert and the search of one task can run on different threads and the search still finds the vector. This holds for committed writes only. It says nothing about a search made while the writing transaction is still open.
512512

513513
If the tail of the search latency matters more than the agent's total time, you can move the rebuild to a moment of your choosing: raise `mutations_before_rebuild` far above the number of writes, set the database setting `arcadedb.vectorIndex.maxDeltaScanRatio` to 0, and call `build_graph_now()` right before a step in which the agent does not search, such as its call to the language model.
514514

0 commit comments

Comments
 (0)