swadhin / justapi public

⚡ p4: native async DB awaits query_async/execute_async (ADR-093) + tests

swadhinbiswas committed 2 months ago
ed4da41
1 parent 5426d7e
commit ed4da41
2 files changed
+148 -0
Showing 2 changed files+148-0
crates/justapi-py/python/justapi/test_native.pycrates/justapi-py/src/database.rs
▼crates/justapi-py/python/justapi/test_native.py
+67-0
@@ -193,3 +193,70 @@ def test_native_async_marker():
193
193
# The decorator marks the raw function; the wrapper propagation is
194
194
# verified by the Rust dispatch using the marker (integration-tested).
195
195
assert getattr(handler, "__native_async__", False) is True
196
+
197
+
198
+def test_native_async_db_query_and_execute():
199
+ """query_async / execute_async: awaitable native DB ops (ADR-093).
200
+
201
+ Runs on a file-based SQLite (in-memory gives each pooled connection its
202
+ own database, which is a test artifact, not a framework bug). Verifies
203
+ both the async query and async write paths return correct results."""
204
+ import os
205
+ import tempfile
206
+ import asyncio
207
+ from justapi import JustAPIApp, JustAPITestClient
208
+
209
+ db_path = os.path.join(tempfile.mkdtemp(), "native_async.db")
210
+ app = JustAPIApp()
211
+ app.set_database(f"sqlite://{db_path}")
212
+ app.connect_database()
213
+ app.db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
214
+ app.db.execute("INSERT INTO users (name) VALUES ('alice'), ('bob')")
215
+
216
+ @app.get("/q")
217
+ async def q(req):
218
+ return await app.db.query_async("SELECT * FROM users ORDER BY id")
219
+
220
+ @app.get("/w")
221
+ async def w(req):
222
+ n = await app.db.execute_async("INSERT INTO users (name) VALUES ('carol')")
223
+ return {"affected": n}
224
+
225
+ c = JustAPITestClient(app)
226
+ r1 = c.get("/q")
227
+ r2 = c.get("/w")
228
+ assert r1["status"] == 200, r1
229
+ assert r2["status"] == 200, r2
230
+ rows = json.loads(r1["body"])
231
+ assert [row["name"] for row in rows] == ["alice", "bob"]
232
+ assert json.loads(r2["body"]) == {"affected": 1}
233
+
234
+ # The write actually persisted (read back through the async path).
235
+ r3 = c.get("/q")
236
+ rows3 = json.loads(r3["body"])
237
+ assert [row["name"] for row in rows3] == ["alice", "bob", "carol"]
238
+
239
+
240
+def test_native_async_db_concurrent_through_http():
241
+ """Many concurrent query_async requests all succeed (parallel DB ops)."""
242
+ import os
243
+ import tempfile
244
+ import concurrent.futures
245
+ from justapi import JustAPIApp, JustAPITestClient
246
+
247
+ db_path = os.path.join(tempfile.mkdtemp(), "native_async_conc.db")
248
+ app = JustAPIApp()
249
+ app.set_database(f"sqlite://{db_path}")
250
+ app.connect_database()
251
+ app.db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
252
+ app.db.execute("INSERT INTO t (v) VALUES ('x')")
253
+
254
+ @app.get("/native")
255
+ async def native(req):
256
+ return await app.db.query_async("SELECT count(*) AS c FROM t")
257
+
258
+ c = JustAPITestClient(app)
259
+ with concurrent.futures.ThreadPoolExecutor(max_workers=16) as ex:
260
+ futs = [ex.submit(c.get, "/native") for _ in range(40)]
261
+ results = [f.result() for f in futs]
262
+ assert all(r["status"] == 200 for r in results), results
196
263
▼crates/justapi-py/src/database.rs
+81-0
@@ -296,6 +296,87 @@ impl DbPool {
296
296
Ok(affected.into_pyobject(py)?.into_any())
297
297
}
298
298
299
+ /// NATIVE ASYNC query (ADR-093): `await`-able from async handlers.
300
+ /// The query runs on the DB's own multi-threaded tokio runtime with the
301
+ /// GIL RELEASED for the whole execution — zero Python stepping during the
302
+ /// query (a native operation type, the one async win our experiments
303
+ /// proved, ADR-090/091). The coroutine suspends on the asyncio loop and
304
+ /// resumes when sqlx completes.
305
+ ///
306
+ /// Usage: `rows = await app.db.query_async("SELECT * FROM users WHERE id = ?", [id])`
307
+ ///
308
+ /// Unlike the blocking `query()` (which must not be called from the loop
309
+ /// thread), `query_async` is safe on the asyncio loop thread and does not
310
+ /// block it.
311
+ #[pyo3(signature = (sql, params=None))]
312
+ fn query_async<'py>(
313
+ &self,
314
+ py: Python<'py>,
315
+ sql: String,
316
+ params: Option<Vec<Py<PyAny>>>,
317
+ ) -> PyResult<Bound<'py, PyAny>> {
318
+ let params_json: Vec<serde_json::Value> = match params {
319
+ Some(ps) => {
320
+ ps.iter().map(|p| python_to_json(py, p.bind(py))).collect::<PyResult<Vec<_>>>()?
321
+ }
322
+ None => Vec::new(),
323
+ };
324
+ let inner = self.inner.clone();
325
+ let rt = self.rt.clone();
326
+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
327
+ let rows: serde_json::Value = rt
328
+ .spawn(async move {
329
+ inner
330
+ .query_with_params(&sql, &params_json)
331
+ .await
332
+ .map_err(|e| anyhow::anyhow!(e.to_string()))
333
+ })
334
+ .await
335
+ .map_err(|e| {
336
+ pyo3::exceptions::PyRuntimeError::new_err(format!("db task join error: {e}"))
337
+ })?
338
+ .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
339
+ // Convert to Python with the GIL (re-attached by the runtime);
340
+ // `.unbind()` yields a Send `Py<PyAny>`.
341
+ pyo3::Python::attach(|py| json_to_python(py, &rows).map(|b| b.unbind()))
342
+ })
343
+ }
344
+
345
+ /// NATIVE ASYNC write (ADR-093): `await`-able INSERT/UPDATE/DELETE/DDL.
346
+ /// Returns the number of rows affected. Same semantics as `query_async`:
347
+ /// runs on the DB tokio runtime, GIL released during execution.
348
+ #[pyo3(signature = (sql, params=None))]
349
+ fn execute_async<'py>(
350
+ &self,
351
+ py: Python<'py>,
352
+ sql: String,
353
+ params: Option<Vec<Py<PyAny>>>,
354
+ ) -> PyResult<Bound<'py, PyAny>> {
355
+ let params_json: Vec<serde_json::Value> = match params {
356
+ Some(ps) => {
357
+ ps.iter().map(|p| python_to_json(py, p.bind(py))).collect::<PyResult<Vec<_>>>()?
358
+ }
359
+ None => Vec::new(),
360
+ };
361
+ let inner = self.inner.clone();
362
+ let rt = self.rt.clone();
363
+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
364
+ let affected: u64 = rt
365
+ .spawn(async move {
366
+ inner
367
+ .execute_with_params(&sql, &params_json)
368
+ .await
369
+ .map_err(|e| anyhow::anyhow!(e.to_string()))
370
+ })
371
+ .await
372
+ .map_err(|e| {
373
+ pyo3::exceptions::PyRuntimeError::new_err(format!("db task join error: {e}"))
374
+ })?
375
+ .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
376
+ Ok(affected)
377
+ })
378
+ }
379
+
299
380
/// Run many `(sql, params)` statements atomically in one transaction and
300
381
/// commit. `stmts` is a list of `[sql, params]` pairs (params optional).
301
382
/// Returns the rows of the final statement if it was a query, else
302
383