@@ -1436,3 +1436,20 @@ runs on the captured tokio `Handle` outside any `Python` token). Throughput is
1436	1436	therefore identical in shape to Step D — bounded by network + pool concurrency,
1437	1437	not by JustAPI. Gates: `cargo test --workspace --features db`, clippy `-D
1438	1438	warnings` on both crates, `cargo fmt --check`, pytest green, Aiven smoke.
	1439	+
	1440	+### P1 follow-up — cross-engine `?` placeholders, `DbParam.bytes`, stream (2026-07-17)
	1441	+
	1442	+`AnyPool` now normalizes `?` → `$N` for Postgres internally, so Python handlers
	1443	+pass `?` unmodified on every engine. Extended smoke (`benchmarks/smoke_dbpool.py`)
	1444	+against live Aiven Postgres now also covers:
	1445	+
	1446	+\| Call \| Result \|
	1447	+\|---\|---\|
	1448	+\| `db.query_stream("SELECT id FROM dbpool_demo ORDER BY id", None, 2)` \| `{"chunks":N, "first":[…]}` (bounded chunks) \|
	1449	+\| `db.execute("INSERT INTO blobs(data) VALUES (?)", [DbParam.bytes(b"\x01\x02\x03")])` \| row inserted (BYTEA) \|
	1450	+\| `db.query("SELECT data FROM blobs …")` \| `[{"data":{"$bytes":"AQID"}}]` (base64 wire → JSON) \|
	1451	+\| `db.transaction([(…),(…)], isolation="serializable")` \| commit, isolation applied for PG/MySQL \|
	1452	+
	1453	+All endpoints return 200 ("DBPOOL SMOKE OK"). Same ~110 ms/round-trip Aiven
	1454	+network bound as before — no added latency. CLI `justapi create <app> --db
	1455	+{postgres,mysql,sqlite}` scaffolds a DB-wired project for each engine (verified).
1439	1456

▼Cargo.lock

+1-0

@@ -2769,6 +2769,7 @@ name = "justapi-py"
2769	2769	version = "2.0.0"
2770	2770	dependencies = [
2771	2771	"anyhow",
	2772	+ "base64 0.22.1",
2772	2773	"futures",
2773	2774	"http",
2774	2775	"http-body",
2775	2776

▼DECISIONS.md

+33-0

@@ -2092,3 +2092,36 @@ Python package does not yet expose a `graphql()` builder, so `/graphql` is never
2092	2092	registered. This is a pre-existing gap unrelated to this change (the core has a
2093	2093	`Handler::GraphQL`, but no Python binding wires it up). Tracked separately.
2094	2094
	2095	+
	2096	+## ADR-056 — 2026-07-17 — Cross-engine `?` placeholder normalization + `justapi create <app>`
	2097	+
	2098	+Context: The `DbPool` public API (`query_with_params` / `execute_with_params`
	2099	+/ `query_stream` / `transaction`) was documented as injection-safe with bound
	2100	+`?` parameters, but Postgres's driver (`sqlx::Any` + Postgres) requires positional
	2101	+`$1`, `$2`, … placeholders. Passing `?` to Postgres produced `syntax error at or
	2102	+near ")"`. SQLite and MySQL accept `?` natively. Without normalization, the same
	2103	+Python code could not run unmodified across all three engines — defeating the
	2104	+"pick any engine at scaffold time" promise of the CLI.
	2105	+
	2106	+Decision / changes:
	2107	+1. `AnyPool::normalize_sql(sql)` (crates/justapi-core/src/db/pool.rs): rewrites
	2108	+ each `?` → `$N` when `self.kind == DbKind::Postgres`; returns the SQL unchanged
	2109	+ for SQLite/MySQL. Applied in `query_with`, `execute_with`, `query_stream`, and
	2110	+ `transaction_with_isolation`. Callers (Python `db.query`/`db.execute`/
	2111	+ `db.query_stream`/`db.transaction`) continue to pass `?` everywhere.
	2112	+2. `justapi create <app>` CLI (crates/justapi-cli/src/main.rs): new
	2113	+ `Commands::Create { name, output, db, db_url }`. `--db {sqlite,postgres,mysql}`
	2114	+ picks the scaffolded backend (default sqlite); `--db-url` overrides the URL.
	2115	+ `resolve_scaffold_db` maps the choice to a default URL; `scaffold_project`
	2116	+ writes a full project (app/main.py wired to `Database`, migrations/, static/,
	2117	+ .env, README, requirements.txt, Dockerfile, .gitignore). `New` now delegates
	2118	+ to `scaffold_project` with sqlite default, removing ~90 lines of duplicated
	2119	+ inline logic.
	2120	+
	2121	+Evidence: `benchmarks/smoke_dbpool.py` extended to cover `query_stream`,
	2122	+`DbParam.bytes` round-trip (`{"$bytes":...}` wire marker → Postgres `BYTEA`), and
	2123	+`transaction`. Run against live Aiven Postgres — all endpoints return 200
	2124	+("DBPOOL SMOKE OK"). Verified `justapi create` scaffolds correct `main.py` for
	2125	+postgres, mysql, and sqlite. `cargo test --workspace --features db` green;
	2126	+`cargo clippy --workspace --tests --features justapi-core/db -D warnings` clean;
	2127	+`cargo fmt --check` clean.
2095	2128

▼benchmarks/smoke_dbpool.py

+19-0

@@ -55,6 +55,21 @@ def test_dbpool_roundtrip():
55	55	ok = db.health()
56	56	return {"ok": ok}
57	57
	58	+ @app.get("/stream")
	59	+ def do_stream(req):
	60	+ db = app.db
	61	+ chunks = db.query_stream("SELECT id FROM dbpool_demo ORDER BY id", None, 2)
	62	+ return {"chunks": len(chunks), "first": chunks[0] if chunks else []}
	63	+
	64	+ @app.get("/blob")
	65	+ def do_blob(req):
	66	+ db = app.db
	67	+ from justapi import DbParam
	68	+ db.execute("CREATE TABLE IF NOT EXISTS blobs (id SERIAL PRIMARY KEY, data BYTEA)")
	69	+ db.execute("INSERT INTO blobs (data) VALUES (?)", [DbParam.bytes(b"\x01\x02\x03")])
	70	+ row = db.query("SELECT data FROM blobs ORDER BY id DESC LIMIT 1")
	71	+ return {"data": row[0]["data"]}
	72	+
58	73	import threading
59	74
60	75	srv = threading.Thread(target=lambda: app.run("127.0.0.1:8123"), daemon=True)
@@ -73,6 +88,10 @@ def test_dbpool_roundtrip():
73	88	txn = get("/txn")
74	89	assert "total" in txn
75	90	assert get("/dbhealth")
	91	+ st = get("/stream")
	92	+ assert "chunks" in st
	93	+ bl = get("/blob")
	94	+ assert "data" in bl
76	95	print("DBPOOL SMOKE OK", get("/seed"))
77	96
78	97
79	98

▼crates/justapi-cli/src/main.rs

+181-96

@@ -152,6 +152,26 @@ enum Commands {
152	152	output: Option<PathBuf>,
153	153	},
154	154
	155	+ /// Create a new JustAPI project with a database backend wired in.
	156	+ ///
	157	+ /// Prompts are avoided: pass `--db` to pick the engine, or `--db-url` to
	158	+ /// supply a full connection string (engine is inferred from the URL scheme).
	159	+ /// Defaults to SQLite (zero-config, file-based).
	160	+ Create {
	161	+ /// Name of the project
	162	+ name: String,
	163	+ /// Output directory (defaults to project name)
	164	+ #[arg(short, long)]
	165	+ output: Option<PathBuf>,
	166	+ /// Database engine: `sqlite` (default), `postgres`, or `mysql`.
	167	+ #[arg(long, default_value = "sqlite")]
	168	+ db: String,
	169	+ /// Full database connection URL. Overrides `--db` (engine is inferred
	170	+ /// from the scheme: `postgres://`, `sqlite://`, `mysql://`).
	171	+ #[arg(long)]
	172	+ db_url: Option<String>,
	173	+ },
	174	+
155	175	/// Profile a running JustAPI server (built-in load test)
156	176	Profile {
157	177	/// Target server address
@@ -270,6 +290,158 @@ fn emit_rich_error(err: &anyhow::Error) {
270	290	}
271	291	}
272	292
	293	+/// Resolve the database engine + connection URL for a scaffolded project.
	294	+///
	295	+/// `--db-url` wins (engine inferred from its scheme); otherwise `--db` picks a
	296	+/// sensible default URL for the chosen engine.
	297	+fn resolve_scaffold_db(db: &str, db_url: Option<String>) -> anyhow::Result<(String, String)> {
	298	+ if let Some(url) = db_url {
	299	+ let kind = if url.starts_with("postgres://") \|\| url.starts_with("postgresql://") {
	300	+ "postgres"
	301	+ } else if url.starts_with("sqlite://") \|\| url.starts_with("sqlite:") {
	302	+ "sqlite"
	303	+ } else if url.starts_with("mysql://") \|\| url.starts_with("mariadb://") {
	304	+ "mysql"
	305	+ } else {
	306	+ anyhow::bail!("Unrecognized database URL scheme: {}", url);
	307	+ };
	308	+ Ok((kind.to_string(), url))
	309	+ } else {
	310	+ let kind = match db.to_ascii_lowercase().as_str() {
	311	+ "postgres" \| "postgresql" => "postgres",
	312	+ "mysql" \| "mariadb" => "mysql",
	313	+ "sqlite" => "sqlite",
	314	+ other => anyhow::bail!(
	315	+ "Unsupported --db '{}'. Choose one of: sqlite, postgres, mysql",
	316	+ other
	317	+ ),
	318	+ };
	319	+ let url = match kind {
	320	+ "postgres" => "postgres://user:pass@localhost:5432/app".to_string(),
	321	+ "mysql" => "mysql://user:pass@localhost:3306/app".to_string(),
	322	+ _ => format!("sqlite://{name}.db", name = "{name}"),
	323	+ };
	324	+ Ok((kind.to_string(), url))
	325	+ }
	326	+}
	327	+
	328	+/// Generate the `app/main.py` for a scaffolded project, wired to the chosen DB.
	329	+fn scaffold_main_py(name: &str, db_kind: &str, db_url: &str) -> String {
	330	+ let db_setup = match db_kind {
	331	+ "postgres" => format!(
	332	+ "app.set_database(\n \"{url}\",\n init_sql=\"\"\"\n CREATE TABLE IF NOT EXISTS items (\n id SERIAL PRIMARY KEY,\n name TEXT NOT NULL,\n qty INT NOT NULL DEFAULT 0\n )\n \"\"\",\n)",
	333	+ url = db_url
	334	+ ),
	335	+ "mysql" => format!(
	336	+ "app.set_database(\n \"{url}\",\n init_sql=\"\"\"\n CREATE TABLE IF NOT EXISTS items (\n id INT AUTO_INCREMENT PRIMARY KEY,\n name VARCHAR(255) NOT NULL,\n qty INT NOT NULL DEFAULT 0\n )\n \"\"\",\n)",
	337	+ url = db_url
	338	+ ),
	339	+ _ => format!(
	340	+ "app.set_database(\n \"{url}\",\n pragmas=[\"journal_mode=WAL\"],\n init_sql=\"\"\"\n CREATE TABLE IF NOT EXISTS items (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n name TEXT NOT NULL,\n qty INTEGER NOT NULL DEFAULT 0\n )\n \"\"\",\n)",
	341	+ url = db_url
	342	+ ),
	343	+ };
	344	+ format!(
	345	+ r#""""JustAPI application — {name}
	346	+
	347	+Database backend: {kind} (URL: {url})
	348	+"""
	349	+from justapi import JustAPIApp, Database
	350	+
	351	+app = JustAPIApp()
	352	+
	353	+# Wire the database. `app.db` is available inside handlers once the server runs.
	354	+{dbsetup}
	355	+
	356	+
	357	+@app.get("/items")
	358	+def list_items(request):
	359	+ # Runs entirely in Rust (GIL released) with bound parameters.
	360	+ return app.db.query("SELECT * FROM items ORDER BY id")
	361	+
	362	+
	363	+@app.post("/items")
	364	+def add_item(request):
	365	+ data = request.json()
	366	+ app.db.execute(
	367	+ "INSERT INTO items (name, qty) VALUES (?, ?)",
	368	+ [data.get("name"), data.get("qty", 0)],
	369	+ )
	370	+ return {{"ok": True}}
	371	+
	372	+
	373	+@app.get("/")
	374	+def root(request):
	375	+ return {{"message": "Hello from {name}!", "db": "{kind}"}}
	376	+"#,
	377	+ name = name,
	378	+ kind = db_kind,
	379	+ url = db_url,
	380	+ dbsetup = db_setup,
	381	+ )
	382	+}
	383	+
	384	+/// Scaffold a new JustAPI project (shared by `new` and `create`).
	385	+fn scaffold_project(
	386	+ name: &str,
	387	+ output: Option<PathBuf>,
	388	+ db_kind: &str,
	389	+ db_url: &str,
	390	+) -> anyhow::Result<()> {
	391	+ let project_dir = output.unwrap_or_else(\|\| PathBuf::from(name));
	392	+ if project_dir.exists() {
	393	+ anyhow::bail!("Directory '{}' already exists", project_dir.display());
	394	+ }
	395	+ std::fs::create_dir_all(&project_dir)?;
	396	+ std::fs::create_dir_all(project_dir.join("app"))?;
	397	+ std::fs::create_dir_all(project_dir.join("migrations"))?;
	398	+ std::fs::create_dir_all(project_dir.join("static"))?;
	399	+
	400	+ std::fs::write(project_dir.join("app").join("__init__.py"), "")?;
	401	+ std::fs::write(
	402	+ project_dir.join("app").join("main.py"),
	403	+ scaffold_main_py(name, db_kind, db_url),
	404	+ )?;
	405	+
	406	+ let env = format!(
	407	+ "# {name} configuration\nHOST=127.0.0.1\nPORT=8080\n# Database engine: {kind}\nDATABASE_URL={url}\n# SECRET_KEY=change-me\n",
	408	+ name = name,
	409	+ kind = db_kind,
	410	+ url = db_url,
	411	+ );
	412	+ std::fs::write(project_dir.join(".env"), env)?;
	413	+
	414	+ std::fs::write(
	415	+ project_dir.join("requirements.txt"),
	416	+ "# Add your Python dependencies here\n# justapi is pre-installed with the runtime\n",
	417	+ )?;
	418	+
	419	+ let readme = format!(
	420	+ "# {name}\n\nA JustAPI application (database backend: {kind}).\n\n## Quick start\n\n```bash\njustapi create {name} --db {kind} # or: justapi new {name}\ncd {dir}\njustapi serve\n```\n\n## Database\n\nConnection: `{url}`\n\n```bash\n# Run migrations\njustapi db migrate --url \"{url}\"\n\n# Start with hot reload\njustapi serve --reload\n```\n",
	421	+ name = name,
	422	+ kind = db_kind,
	423	+ dir = project_dir.display(),
	424	+ url = db_url,
	425	+ );
	426	+ std::fs::write(project_dir.join("README.md"), readme)?;
	427	+
	428	+ std::fs::write(
	429	+ project_dir.join(".gitignore"),
	430	+ ".pyc\n__pycache__/\n.env\n.egg-info/\ndist/\nbuild/\n*.db\n",
	431	+ )?;
	432	+
	433	+ std::fs::write(
	434	+ project_dir.join("Dockerfile"),
	435	+ "FROM python:3.12-slim\nWORKDIR /app\nCOPY requirements.txt .\nRUN pip install justapi\nCOPY . .\nEXPOSE 8080\nCMD [\"justapi\", \"serve\", \"--host\", \"0.0.0.0\", \"--port\", \"8080\"]\n",
	436	+ )?;
	437	+
	438	+ println!("✅ Created new JustAPI project '{}' (database: {})", name, db_kind);
	439	+ println!();
	440	+ println!(" cd {}", project_dir.display());
	441	+ println!(" justapi serve");
	442	+ Ok(())
	443	+}
	444	+
273	445	async fn run() -> anyhow::Result<()> {
274	446	justapi_core::tracing_setup::init_tracing()?;
275	447
@@ -530,6 +702,7 @@ async fn run() -> anyhow::Result<()> {
530	702	kind: None,
531	703	init_sql: None,
532	704	pragmas: None,
	705	+ ..Default::default()
533	706	};
534	707	let mut mgr = justapi_core::db::PoolManager::new();
535	708	let pool = mgr.init("", config).await?;
@@ -553,6 +726,7 @@ async fn run() -> anyhow::Result<()> {
553	726	kind: None,
554	727	init_sql: None,
555	728	pragmas: None,
	729	+ ..Default::default()
556	730	};
557	731	let mut mgr = justapi_core::db::PoolManager::new();
558	732	let pool = mgr.init("", config).await?;
@@ -570,6 +744,7 @@ async fn run() -> anyhow::Result<()> {
570	744	kind: None,
571	745	init_sql: None,
572	746	pragmas: None,
	747	+ ..Default::default()
573	748	};
574	749	let mut mgr = justapi_core::db::PoolManager::new();
575	750	let pool = mgr.init("", config).await?;
@@ -725,102 +900,12 @@ async fn run() -> anyhow::Result<()> {
725	900	Ok(())
726	901	}
727	902	Commands::New { name, output } => {
728		- let project_dir = output.unwrap_or_else(\|\| PathBuf::from(&name));
729		- if project_dir.exists() {
730		- anyhow::bail!("Directory '{}' already exists", project_dir.display());
731		- }
732		- std::fs::create_dir_all(&project_dir)?;
733		- std::fs::create_dir_all(project_dir.join("app"))?;
734		- std::fs::create_dir_all(project_dir.join("migrations"))?;
735		- std::fs::create_dir_all(project_dir.join("static"))?;
736		-
737		- // Create main.py
738		- let main_py = format!(
739		- r#""""JustAPI application — {}
740		-"""
741		-
742		-from justapi import JustAPIApp
743		-
744		-app = JustAPIApp()
745		-
746		-
747		-@app.get("/")
748		-async def root():
749		- return {{"message": "Hello from {}!"}}
750		-
751		-
752		-@app.get("/health")
753		-async def health():
754		- return {{"status": "healthy"}}
755		-
756		-
757		-if __name__ == "__main__":
758		- app.run()
759		-"#,
760		- name, name
761		- );
762		- std::fs::write(project_dir.join("app").join("__init__.py"), "")?;
763		- std::fs::write(project_dir.join("app").join("main.py"), &main_py)?;
764		-
765		- // Create .env
766		- let env = format!(
767		- r#"# {} configuration
768		-HOST=127.0.0.1
769		-PORT=8080
770		-# DATABASE_URL=postgres://user:pass@localhost:5432/{}
771		-# SECRET_KEY=change-me
772		-"#,
773		- name, name
774		- );
775		- std::fs::write(project_dir.join(".env"), &env)?;
776		-
777		- // Create requirements.txt
778		- std::fs::write(
779		- project_dir.join("requirements.txt"),
780		- "# Add your Python dependencies here\n# justapi is pre-installed with the runtime\n",
781		- )?;
782		-
783		- // Create README.md
784		- let readme = format!(
785		- r#"# {}
786		-
787		-A JustAPI application.
788		-
789		-## Quick start
790		-
791		-```bash
792		-justapi serve
793		-```
794		-
795		-## Development
796		-
797		-```bash
798		-# Install dependencies
799		-pip install -r requirements.txt
800		-
801		-# Run migrations
802		-justapi db migrate --url postgres://localhost/{}
803		-
804		-# Start server with hot reload
805		-justapi serve --reload
806		-```
807		-"#,
808		- name, name
809		- );
810		- std::fs::write(project_dir.join("README.md"), &readme)?;
811		-
812		- // Create .gitignore
813		- std::fs::write(
814		- project_dir.join(".gitignore"),
815		- ".pyc\n__pycache__/\n.env\n.egg-info/\ndist/\nbuild/\n",
816		- )?;
817		-
818		- println!("✅ Created new JustAPI project '{}'", name);
819		- println!();
820		- println!(" cd {}", project_dir.display());
821		- println!(" justapi serve");
822		-
823		- Ok(())
	903	+ // Default scaffold: SQLite (zero-config).
	904	+ scaffold_project(&name, output, "sqlite", "sqlite://app.db")
	905	+ }
	906	+ Commands::Create { name, output, db, db_url } => {
	907	+ let (kind, url) = resolve_scaffold_db(&db, db_url)?;
	908	+ scaffold_project(&name, output, &kind, &url)
824	909	}
825	910	Commands::Profile { addr, duration, connections, output } => {
826	911	println!("🔬 JustAPI Profiler");
827	912

▼crates/justapi-core/Cargo.toml

+2-2

@@ -7,7 +7,7 @@ license.workspace = true
7	7	[features]
8	8	default = ["opentelemetry"]
9	9	tls = ["rustls", "tokio-rustls", "rustls-pemfile"]
10		-ws = ["tokio-tungstenite", "sha1", "base64"]
	10	+ws = ["tokio-tungstenite", "sha1"]
11	11	compression = []
12	12	brotli-compression = ["compression", "brotli"]
13	13	zstd-compression = ["compression", "zstd"]
@@ -45,7 +45,7 @@ simd-json = { workspace = true, optional = true }
45	45	tokio-tungstenite = { workspace = true, optional = true }
46	46	tokio-stream.workspace = true
47	47	sha1 = { workspace = true, optional = true }
48		-base64 = { workspace = true, optional = true }
	48	+base64 = { workspace = true }
49	49	jsonschema.workspace = true
50	50	rustls = { workspace = true, optional = true }
51	51	tokio-rustls = { workspace = true, optional = true }
52	52

▼crates/justapi-core/src/db/pool.rs

+470-76

@@ -6,7 +6,11 @@
6	6	use std::collections::HashMap;
7	7	use std::fmt;
8	8	use std::sync::Arc;
	9	+use std::time::Duration;
9	10
	11	+use base64::engine::general_purpose::STANDARD as B64;
	12	+use base64::Engine;
	13	+use futures::TryStreamExt as _;
10	14	use sqlx::any::AnyPoolOptions;
11	15	use sqlx::{Column, Executor, Row};
12	16	use tokio::sync::RwLock;
@@ -33,9 +37,9 @@ impl DbKind {
33	37	pub fn from_url(url: &str) -> Self {
34	38	if url.starts_with("postgres://") \|\| url.starts_with("postgresql://") {
35	39	DbKind::Postgres
36		- } else if url.starts_with("sqlite://") {
	40	+ } else if url.starts_with("sqlite://") \|\| url.starts_with("sqlite:") {
37	41	DbKind::Sqlite
38		- } else if url.starts_with("mysql://") {
	42	+ } else if url.starts_with("mysql://") \|\| url.starts_with("mariadb://") {
39	43	DbKind::MySql
40	44	} else {
41	45	DbKind::Postgres
@@ -43,6 +47,40 @@ impl DbKind {
43	47	}
44	48	}
45	49
	50	+/// Transaction isolation level, mapped to each engine's dialect.
	51	+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
	52	+pub enum IsolationLevel {
	53	+ ReadUncommitted,
	54	+ ReadCommitted,
	55	+ RepeatableRead,
	56	+ Serializable,
	57	+ /// Snapshot isolation (Postgres `REPEATABLE READ`/`SERIALIZABLE` semantics
	58	+ /// vary; this maps to the engine's default repeatable/snapshot level).
	59	+ Snapshot,
	60	+}
	61	+
	62	+impl IsolationLevel {
	63	+ /// The `SET TRANSACTION ISOLATION LEVEL ...` fragment for this engine.
	64	+ pub fn to_sql(self, _kind: DbKind) -> &'static str {
	65	+ match self {
	66	+ IsolationLevel::ReadUncommitted => "READ UNCOMMITTED",
	67	+ IsolationLevel::ReadCommitted => "READ COMMITTED",
	68	+ IsolationLevel::RepeatableRead => "REPEATABLE READ",
	69	+ IsolationLevel::Serializable => "SERIALIZABLE",
	70	+ IsolationLevel::Snapshot => "REPEATABLE READ",
	71	+ }
	72	+ }
	73	+
	74	+ pub fn is_supported(self, kind: DbKind) -> bool {
	75	+ // SQLite only supports SERIALIZABLE (its default). Everything else
	76	+ // honours the standard levels.
	77	+ match kind {
	78	+ DbKind::Sqlite => self == IsolationLevel::Serializable,
	79	+ _ => true,
	80	+ }
	81	+ }
	82	+}
	83	+
46	84	/// Translate a user-facing DB URL into the form `sqlx::Any` expects.
47	85	///
48	86	/// `sqlx::Any` is stricter than the per-driver pools about URL shape: SQLite
@@ -71,19 +109,99 @@ pub struct DatabaseConfig {
71	109	pub max_connections: u32,
72	110	pub kind: Option<DbKind>,
73	111	/// Optional DDL run once right after the pool is created (e.g. `CREATE
74		- /// TABLE ...`). Bounded by a single statement; used by the Python app to
75		- /// bootstrap schema for Rust-native CRUD routes.
	112	+ /// TABLE ...`). Multiple `;`-separated statements are each run in order.
76	113	pub init_sql: Option<String>,
77	114	/// Optional PRAGMA / session statements run on every connection right after
78	115	/// it is checked out of the pool (e.g. `journal_mode=WAL`,
79	116	/// `synchronous=NORMAL`). SQLite-only; ignored by other drivers. Applied via
80	117	/// `after_connect` so every pooled connection picks them up.
81	118	pub pragmas: Option<Vec<String>>,
	119	+ /// Maximum time to wait for a connection from the pool (acquire timeout).
	120	+ /// `None` uses sqlx's default (no timeout).
	121	+ pub acquire_timeout: Option<Duration>,
	122	+ /// Maximum idle time before a connection is closed and recycled.
	123	+ pub idle_timeout: Option<Duration>,
	124	+ /// Maximum lifetime of a connection before it is forcibly closed.
	125	+ pub max_lifetime: Option<Duration>,
	126	+ /// Background health-check interval. When `Some`, the [`PoolManager`] spawns
	127	+ /// a task that pings the pool and logs/marks it unhealthy. `None` disables
	128	+ /// the loop.
	129	+ pub health_check_interval: Option<Duration>,
	130	+ /// Default transaction isolation level for `begin()`/`transaction()`.
	131	+ pub default_isolation: Option<IsolationLevel>,
82	132	}
83	133
84	134	impl Default for DatabaseConfig {
85	135	fn default() -> Self {
86		- Self { url: String::new(), max_connections: 10, kind: None, init_sql: None, pragmas: None }
	136	+ Self {
	137	+ url: String::new(),
	138	+ max_connections: 10,
	139	+ kind: None,
	140	+ init_sql: None,
	141	+ pragmas: None,
	142	+ acquire_timeout: Some(Duration::from_secs(30)),
	143	+ idle_timeout: None,
	144	+ max_lifetime: Some(Duration::from_secs(1800)),
	145	+ health_check_interval: None,
	146	+ default_isolation: None,
	147	+ }
	148	+ }
	149	+}
	150	+
	151	+/// A typed parameter value for a prepared statement.
	152	+///
	153	+/// `serde_json::Value` is the convenience path, but it cannot represent raw
	154	+/// blobs or full-precision decimals. [`Param`] adds those cases so handlers can
	155	+/// bind `bytes`/`BLOB` columns and pass native JSON directly.
	156	+#[derive(Debug, Clone)]
	157	+pub enum Param {
	158	+ Null,
	159	+ Bool(bool),
	160	+ Int(i64),
	161	+ Float(f64),
	162	+ Text(String),
	163	+ /// Raw bytes, bound as a BLOB. Round-trips back as `{"$bytes": "<base64>"}`.
	164	+ Bytes(Vec<u8>),
	165	+ /// A JSON document, bound as a native JSON/JSONB value where the driver
	166	+ /// supports it, else serialized to text.
	167	+ Json(serde_json::Value),
	168	+}
	169	+
	170	+impl From<&serde_json::Value> for Param {
	171	+ fn from(v: &serde_json::Value) -> Self {
	172	+ // Typed-param wire markers emitted by the Python bridge. `DbParam.bytes`
	173	+ // serializes to `{"$bytes": "<base64>"}` so BLOB values survive the
	174	+ // JSON bridge and are bound as a real BLOB (not text).
	175	+ if let Some(obj) = v.as_object() {
	176	+ if let Some(b64) = obj.get("$bytes").and_then(\|x\| x.as_str()) {
	177	+ return match B64.decode(b64) {
	178	+ Ok(b) => Param::Bytes(b),
	179	+ Err(_) => Param::Text(b64.to_string()),
	180	+ };
	181	+ }
	182	+ }
	183	+ match v {
	184	+ serde_json::Value::Null => Param::Null,
	185	+ serde_json::Value::Bool(b) => Param::Bool(*b),
	186	+ serde_json::Value::Number(n) => {
	187	+ if let Some(i) = n.as_i64() {
	188	+ Param::Int(i)
	189	+ } else if let Some(f) = n.as_f64() {
	190	+ Param::Float(f)
	191	+ } else {
	192	+ Param::Text(n.to_string())
	193	+ }
	194	+ }
	195	+ serde_json::Value::String(s) => Param::Text(s.clone()),
	196	+ other => Param::Json(other.clone()),
	197	+ }
	198	+ }
	199	+}
	200	+
	201	+impl Param {
	202	+ /// Convert a slice of JSON values into typed params (the common case).
	203	+ pub fn from_json_slice(vals: &[serde_json::Value]) -> Vec<Param> {
	204	+ vals.iter().map(Param::from).collect()
87	205	}
88	206	}
89	207
@@ -98,6 +216,7 @@ impl Default for DatabaseConfig {
98	216	pub struct AnyPool {
99	217	inner: sqlx::AnyPool,
100	218	kind: DbKind,
	219	+ default_isolation: Option<IsolationLevel>,
101	220	}
102	221
103	222	impl AnyPool {
@@ -111,12 +230,26 @@ impl AnyPool {
111	230	kind: DbKind,
112	231	pragmas: Option<Vec<String>>,
113	232	) -> Result<Self, sqlx::Error> {
114		- // The `any` driver needs its backend drivers registered before the first
115		- // connection (sqlx 0.8). Idempotent and cheap.
	233	+ Self::connect(config(url, max_connections, kind, pragmas)).await
	234	+ }
	235	+
	236	+ /// Connect from a full [`DatabaseConfig`], applying pool tuning (acquire
	237	+ /// timeout, idle/lifetime recycling) and SQLite PRAGMAs.
	238	+ pub async fn connect(config: DatabaseConfig) -> Result<Self, sqlx::Error> {
	239	+ let kind = config.kind.unwrap_or_else(\|\| DbKind::from_url(&config.url));
116	240	sqlx::any::install_default_drivers();
117		- let connect_url = normalize_db_url(url);
118		- let mut opts = AnyPoolOptions::new().max_connections(max_connections);
119		- if let (DbKind::Sqlite, Some(pragmas)) = (kind, pragmas) {
	241	+ let connect_url = normalize_db_url(&config.url);
	242	+ let mut opts = AnyPoolOptions::new().max_connections(config.max_connections);
	243	+ if let Some(t) = config.acquire_timeout {
	244	+ opts = opts.acquire_timeout(t);
	245	+ }
	246	+ if let Some(t) = config.idle_timeout {
	247	+ opts = opts.idle_timeout(t);
	248	+ }
	249	+ if let Some(t) = config.max_lifetime {
	250	+ opts = opts.max_lifetime(t);
	251	+ }
	252	+ if let (DbKind::Sqlite, Some(pragmas)) = (kind, config.pragmas) {
120	253	opts = opts.after_connect(move \|conn, _meta\| {
121	254	let pragmas = pragmas.clone();
122	255	Box::pin(async move {
@@ -128,13 +261,38 @@ impl AnyPool {
128	261	});
129	262	}
130	263	let inner = opts.connect(&connect_url).await?;
131		- Ok(Self { inner, kind })
	264	+ Ok(Self { inner, kind, default_isolation: config.default_isolation })
132	265	}
133	266
134	267	pub fn kind(&self) -> DbKind {
135	268	self.kind
136	269	}
137	270
	271	+ /// Normalize `?` placeholders to the driver's native style. Postgres uses
	272	+ /// positional `$1`, `$2`, ...; SQLite and MySQL accept `?` natively so the
	273	+ /// SQL is returned unchanged. This lets callers always pass `?` regardless
	274	+ /// of the backing engine.
	275	+ fn normalize_sql(&self, sql: &str) -> String {
	276	+ if self.kind != DbKind::Postgres {
	277	+ return sql.to_string();
	278	+ }
	279	+ let mut out = String::with_capacity(sql.len());
	280	+ let mut idx: usize = 0;
	281	+ for ch in sql.chars() {
	282	+ if ch == '?' {
	283	+ idx += 1;
	284	+ out.push_str(&format!("${}", idx));
	285	+ } else {
	286	+ out.push(ch);
	287	+ }
	288	+ }
	289	+ out
	290	+ }
	291	+
	292	+ pub fn default_isolation(&self) -> Option<IsolationLevel> {
	293	+ self.default_isolation
	294	+ }
	295	+
138	296	/// Access the underlying type-erased pool.
139	297	pub fn as_any(&self) -> &sqlx::AnyPool {
140	298	&self.inner
@@ -160,24 +318,39 @@ impl AnyPool {
160	318	Ok(rows.into_iter().map(\|r\| r.0).collect())
161	319	}
162	320
163		- /// Begin a transaction.
	321	+ /// Begin a transaction with the pool's default isolation level (if set).
164	322	pub async fn begin(&self) -> Result<sqlx::Transaction<'static, sqlx::Any>, sqlx::Error> {
165		- self.inner.begin().await
	323	+ self.begin_with(self.default_isolation).await
	324	+ }
	325	+
	326	+ /// Begin a transaction with an explicit isolation level. If the engine does
	327	+ /// not support the requested level (e.g. non-SERIALIZABLE on SQLite), it
	328	+ /// falls back to the engine default rather than erroring.
	329	+ pub async fn begin_with(
	330	+ &self,
	331	+ isolation: Option<IsolationLevel>,
	332	+ ) -> Result<sqlx::Transaction<'static, sqlx::Any>, sqlx::Error> {
	333	+ let mut tx = self.inner.begin().await?;
	334	+ if let Some(iso) = isolation {
	335	+ if iso.is_supported(self.kind) && self.kind != DbKind::Sqlite {
	336	+ let stmt = format!("SET TRANSACTION ISOLATION LEVEL {}", iso.to_sql(self.kind));
	337	+ tx.execute(sqlx::query(&stmt)).await?;
	338	+ }
	339	+ }
	340	+ Ok(tx)
166	341	}
167	342
168	343	/// Run a query and return all rows as a JSON array of objects, mirroring how
169	344	/// the Python bridge returns query results. Used by Rust-native CRUD routes
170	345	/// so a `INSERT ... RETURNING` / `SELECT` can be served without touching the
171		- /// GIL. Column values are mapped best-effort: integers, floats, booleans,
172		- /// strings, and NULL become their JSON equivalents; other types fall back to
173		- /// their string form.
	346	+ /// GIL.
174	347	pub async fn query_json(&self, sql: &str) -> Result<serde_json::Value, sqlx::Error> {
175	348	let rows = sqlx::query(sql).fetch_all(&self.inner).await?;
176	349	rows_to_json(rows)
177	350	}
178	351
179	352	/// Run a SQL statement with bound parameters and return all rows as a JSON
180		- /// array of objects. Parameters are downcast from JSON via [`bind_json`], so
	353	+ /// array of objects. Parameters are downcast from JSON via [`bind_param`], so
181	354	/// this is injection-safe (no string interpolation). Used by the Rust-native
182	355	/// CRUD handlers (select/update/delete) which build the SQL string but bind
183	356	/// every value as a parameter. Also exposed to Python handlers via the
@@ -187,9 +360,20 @@ impl AnyPool {
187	360	sql: &str,
188	361	params: &[serde_json::Value],
189	362	) -> Result<serde_json::Value, sqlx::Error> {
190		- let mut q = sqlx::query(sql);
191		- for v in params {
192		- q = bind_json(q, v);
	363	+ self.query_with(sql, &Param::from_json_slice(params)).await
	364	+ }
	365	+
	366	+ /// Like [`AnyPool::query_with_params`] but accepts typed [`Param`]s (supports
	367	+ /// BLOB binding and native JSON values).
	368	+ pub async fn query_with(
	369	+ &self,
	370	+ sql: &str,
	371	+ params: &[Param],
	372	+ ) -> Result<serde_json::Value, sqlx::Error> {
	373	+ let sql = self.normalize_sql(sql);
	374	+ let mut q = sqlx::query(&sql);
	375	+ for p in params {
	376	+ q = bind_param(q, p);
193	377	}
194	378	let rows = q.fetch_all(&self.inner).await?;
195	379	rows_to_json(rows)
@@ -202,46 +386,119 @@ impl AnyPool {
202	386	sql: &str,
203	387	params: &[serde_json::Value],
204	388	) -> Result<u64, sqlx::Error> {
205		- let mut q = sqlx::query(sql);
206		- for v in params {
207		- q = bind_json(q, v);
	389	+ self.execute_with(sql, &Param::from_json_slice(params)).await
	390	+ }
	391	+
	392	+ /// Like [`AnyPool::execute_with_params`] but accepts typed [`Param`]s.
	393	+ pub async fn execute_with(&self, sql: &str, params: &[Param]) -> Result<u64, sqlx::Error> {
	394	+ let sql = self.normalize_sql(sql);
	395	+ let mut q = sqlx::query(&sql);
	396	+ for p in params {
	397	+ q = bind_param(q, p);
208	398	}
209	399	let res = q.execute(&self.inner).await?;
210	400	Ok(res.rows_affected())
211	401	}
212	402
	403	+ /// Stream a query in batches of `chunk` rows, returning a `Vec` of row-chunks.
	404	+ ///
	405	+ /// Unlike [`AnyPool::query_with`] (which buffers the whole result set), this
	406	+ /// reads the result set incrementally and yields each `chunk`-sized window as
	407	+ /// a JSON array, so a very large result set stays bounded in memory on the
	408	+ /// Rust side. Each chunk is itself a JSON array of row objects.
	409	+ pub async fn query_stream(
	410	+ &self,
	411	+ sql: &str,
	412	+ params: &[Param],
	413	+ chunk: usize,
	414	+ ) -> Result<Vec<serde_json::Value>, sqlx::Error> {
	415	+ let chunk = chunk.max(1);
	416	+ let sql = self.normalize_sql(sql);
	417	+ let mut q = sqlx::query(&sql);
	418	+ for p in params {
	419	+ q = bind_param(q, p);
	420	+ }
	421	+ let mut rows = q.fetch(&self.inner);
	422	+ let mut chunks: Vec<serde_json::Value> = Vec::new();
	423	+ let mut buf: Vec<serde_json::Value> = Vec::with_capacity(chunk);
	424	+ while let Some(row) = rows.try_next().await? {
	425	+ buf.push(row_to_json(&row)?);
	426	+ if buf.len() == chunk {
	427	+ chunks.push(serde_json::Value::Array(std::mem::take(&mut buf)));
	428	+ }
	429	+ }
	430	+ if !buf.is_empty() {
	431	+ chunks.push(serde_json::Value::Array(buf));
	432	+ }
	433	+ Ok(chunks)
	434	+ }
	435	+
213	436	/// Run a list of `(sql, params)` statements atomically inside one transaction
214	437	/// and commit. On any error the transaction is rolled back. Returns the rows
215		- /// of the final statement (as JSON) if it was a query, else an object with the
216		- /// total rows affected. This is the production-safe multi-statement primitive
217		- /// exposed to Python as `DbPool.transaction([...])`.
	438	+ /// of the final statement (as JSON) if it produced rows, else an object with
	439	+ /// the total rows affected.
	440	+ ///
	441	+ /// Row-vs-write detection is robust: each statement is fetched; if it
	442	+ /// returns rows they are recorded, otherwise its `rows_affected` is folded
	443	+ /// into the running total. This handles `WITH ... SELECT`, `WITH ... INSERT`,
	444	+ /// and `RETURNING` correctly without brittle keyword probing.
218	445	pub async fn transaction(
219	446	&self,
220	447	stmts: &[(String, Vec<serde_json::Value>)],
221	448	) -> Result<serde_json::Value, sqlx::Error> {
222		- let mut tx = self.inner.begin().await?;
	449	+ let typed: Vec<(String, Vec<Param>)> =
	450	+ stmts.iter().map(\|(s, p)\| (s.clone(), Param::from_json_slice(p))).collect();
	451	+ self.transaction_with(&typed).await
	452	+ }
	453	+
	454	+ /// Like [`AnyPool::transaction`] but accepts typed [`Param`]s.
	455	+ pub async fn transaction_with(
	456	+ &self,
	457	+ stmts: &[(String, Vec<Param>)],
	458	+ ) -> Result<serde_json::Value, sqlx::Error> {
	459	+ self.transaction_with_isolation(stmts, self.default_isolation).await
	460	+ }
	461	+
	462	+ /// Run a transaction with an explicit isolation level (overriding the pool
	463	+ /// default for this call). See [`AnyPool::transaction_with`].
	464	+ pub async fn transaction_with_isolation(
	465	+ &self,
	466	+ stmts: &[(String, Vec<Param>)],
	467	+ isolation: Option<IsolationLevel>,
	468	+ ) -> Result<serde_json::Value, sqlx::Error> {
	469	+ let mut tx = self.begin_with(isolation).await?;
223	470	let mut last_rows: serde_json::Value = serde_json::Value::Null;
224	471	let mut total_affected: u64 = 0;
225	472	for (sql, params) in stmts {
226		- let mut q = sqlx::query(sql);
227		- for v in params {
228		- q = bind_json(q, v);
	473	+ let sql = self.normalize_sql(sql);
	474	+ let mut q = sqlx::query(&sql);
	475	+ for p in params {
	476	+ q = bind_param(q, p);
229	477	}
230		- // Try to fetch (SELECT) first; if it's not a query, it returns no rows
231		- // and we fall back to execute (rows affected). `sqlx::query` supports
232		- // both, but `fetch_all` on a non-SELECT still works (0 rows). We use
233		- // `execute` for writes and `fetch_all` for reads by probing the SQL.
234		- let upper = sql.trim_start().to_uppercase();
235		- if upper.starts_with("SELECT")
236		- \|\| upper.starts_with("WITH")
237		- \|\| upper.starts_with("RETURNING")
238		- {
239		- let rows = q.fetch_all(&mut *tx).await?;
240		- last_rows = rows_to_json(rows)?;
241		- } else {
242		- let res = q.execute(&mut *tx).await?;
243		- total_affected += res.rows_affected();
	478	+ // `fetch_many` yields each row (`Left`) and, once, the final
	479	+ // `QueryResult` (`Right`) which carries `rows_affected`. This
	480	+ // executes the statement exactly once and simultaneously tells us
	481	+ // whether it was a read (rows present) or a write (rows_affected),
	482	+ // without brittle keyword probing and without re-executing.
	483	+ #[allow(deprecated)]
	484	+ let mut stream = q.fetch_many(&mut *tx);
	485	+ let mut rows: Vec<serde_json::Value> = Vec::new();
	486	+ let mut affected: u64 = 0;
	487	+ while let Some(item) = stream.try_next().await? {
	488	+ match item {
	489	+ sqlx::Either::Left(result) => {
	490	+ affected = result.rows_affected();
	491	+ }
	492	+ sqlx::Either::Right(row) => {
	493	+ rows.push(row_to_json(&row)?);
	494	+ }
	495	+ }
	496	+ }
	497	+ if rows.is_empty() {
	498	+ total_affected += affected;
244	499	last_rows = serde_json::Value::Null;
	500	+ } else {
	501	+ last_rows = serde_json::Value::Array(rows);
245	502	}
246	503	}
247	504	tx.commit().await?;
@@ -273,11 +530,11 @@ impl AnyPool {
273	530	sqlx::Error::ColumnNotFound("request body must be a JSON object".into())
274	531	})?;
275	532	let mut cols: Vec<String> = Vec::new();
276		- let mut vals: Vec<&serde_json::Value> = Vec::new();
	533	+ let mut vals: Vec<Param> = Vec::new();
277	534	for c in columns {
278	535	if let Some(v) = obj.get(c) {
279	536	cols.push(c.clone());
280		- vals.push(v);
	537	+ vals.push(Param::from(v));
281	538	}
282	539	}
283	540	if cols.is_empty() {
@@ -298,34 +555,26 @@ impl AnyPool {
298	555	);
299	556	let mut q = sqlx::query(&sql);
300	557	for v in &vals {
301		- q = bind_json(q, v);
	558	+ q = bind_param(q, v);
302	559	}
303	560	let rows = q.fetch_all(&self.inner).await?;
304	561	rows_to_json(rows)
305	562	}
306	563	}
307	564
308		-/// Bind a JSON value as a query parameter for `sqlx::Any`. Scalars are downcast
309		-/// to the concrete Rust type sqlx expects (`i64`/`f64`/`bool`/`String`); JSON
310		-/// objects/arrays are stringified. `sqlx::Any` accepts `Option<String>` for NULL.
311		-fn bind_json<'q>(
	565	+/// Bind a [`Param`] as a query parameter for `sqlx::Any`.
	566	+fn bind_param<'q>(
312	567	q: sqlx::query::Query<'q, sqlx::Any, <sqlx::Any as sqlx::Database>::Arguments<'q>>,
313		- v: &serde_json::Value,
	568	+ v: &Param,
314	569	) -> sqlx::query::Query<'q, sqlx::Any, <sqlx::Any as sqlx::Database>::Arguments<'q>> {
315	570	match v {
316		- serde_json::Value::Null => q.bind(Option::<String>::None),
317		- serde_json::Value::Bool(b) => q.bind(*b),
318		- serde_json::Value::Number(n) => {
319		- if let Some(i) = n.as_i64() {
320		- q.bind(i)
321		- } else if let Some(f) = n.as_f64() {
322		- q.bind(f)
323		- } else {
324		- q.bind(n.to_string())
325		- }
326		- }
327		- serde_json::Value::String(s) => q.bind(s.clone()),
328		- other => q.bind(other.to_string()),
	571	+ Param::Null => q.bind(Option::<String>::None),
	572	+ Param::Bool(b) => q.bind(*b),
	573	+ Param::Int(i) => q.bind(*i),
	574	+ Param::Float(f) => q.bind(*f),
	575	+ Param::Text(s) => q.bind(s.clone()),
	576	+ Param::Bytes(b) => q.bind(b.clone()),
	577	+ Param::Json(j) => q.bind(j.to_string()),
329	578	}
330	579	}
331	580
@@ -344,27 +593,88 @@ fn rows_to_json(rows: Vec<sqlx::any::AnyRow>) -> Result<serde_json::Value, sqlx:
344	593	Ok(serde_json::Value::Array(out))
345	594	}
346	595
347		-/// Best-effort mapping of a single `sqlx::any::AnyRow` column value to JSON.
	596	+/// Convert a single `sqlx::any::AnyRow` into a JSON object (column-keyed).
	597	+fn row_to_json(row: &sqlx::any::AnyRow) -> Result<serde_json::Value, sqlx::Error> {
	598	+ let mut obj = serde_json::Map::new();
	599	+ for col in row.columns() {
	600	+ let name = col.name().to_string();
	601	+ let val: serde_json::Value = column_to_json(row, col.ordinal())?;
	602	+ obj.insert(name, val);
	603	+ }
	604	+ Ok(serde_json::Value::Object(obj))
	605	+}
	606	+
	607	+/// Best-effort, fidelity-preserving mapping of a single `sqlx::any::AnyRow`
	608	+/// column value to JSON.
	609	+///
	610	+/// Mapping rules (in priority order):
	611	+/// - NULL → `null`
	612	+/// - `bool` → JSON bool
	613	+/// - 32/64-bit integers → JSON integer (kept as `i64` to preserve precision)
	614	+/// - `f64` → JSON number, but non-finite values (NaN/±inf) become `null`
	615	+/// because JSON has no representation for them (prevents serialization panics)
	616	+/// - `Vec<u8>` (BLOB) → `{"$bytes": "<base64>"}` so the client can detect and
	617	+/// decode binary without ambiguity
	618	+/// - everything else (text, UUID, DECIMAL, TIMESTAMP, JSONB, enums, …) → JSON
	619	+/// string (the canonical text form). High-precision DECIMAL is preserved as a
	620	+/// string rather than lossily coerced to `f64`.
348	621	fn column_to_json(row: &sqlx::any::AnyRow, idx: usize) -> Result<serde_json::Value, sqlx::Error> {
	622	+ // NULL check first (try_get::<Option<_>> yields None for NULL).
349	623	if let Ok(v) = row.try_get::<Option<bool>, _>(idx) {
350	624	return Ok(serde_json::Value::from(v));
351	625	}
	626	+ if let Ok(v) = row.try_get::<Option<i32>, _>(idx) {
	627	+ return Ok(serde_json::Value::from(v.map(i64::from)));
	628	+ }
352	629	if let Ok(v) = row.try_get::<Option<i64>, _>(idx) {
353	630	return Ok(serde_json::Value::from(v));
354	631	}
355	632	if let Ok(v) = row.try_get::<Option<f64>, _>(idx) {
356		- return Ok(serde_json::Value::from(v));
	633	+ // Guard against NaN/inf: JSON cannot represent them.
	634	+ return Ok(match v {
	635	+ Some(f) if f.is_finite() => serde_json::Value::from(f),
	636	+ _ => serde_json::Value::Null,
	637	+ });
	638	+ }
	639	+ // BLOB / bytea → base64-wrapped object so binary survives the JSON bridge.
	640	+ if let Ok(v) = row.try_get::<Option<Vec<u8>>, _>(idx) {
	641	+ return Ok(match v {
	642	+ Some(b) => serde_json::json!({ "$bytes": B64.encode(b) }),
	643	+ None => serde_json::Value::Null,
	644	+ });
357	645	}
358	646	if let Ok(v) = row.try_get::<Option<String>, _>(idx) {
359	647	return Ok(serde_json::Value::from(v));
360	648	}
361		- // Fallback: render as string (covers blobs, dates, decimals, etc.).
	649	+ // Fallback: render as string (covers dates, decimals, JSONB text, etc.).
362	650	match row.try_get::<Option<String>, _>(idx) {
363	651	Ok(s) => Ok(serde_json::Value::from(s)),
364	652	Err(e) => Err(e),
365	653	}
366	654	}
367	655
	656	+/// Build a [`DatabaseConfig`] from the common fields (kept for callers that do
	657	+/// not need pool tuning).
	658	+fn config(
	659	+ url: &str,
	660	+ max_connections: u32,
	661	+ kind: DbKind,
	662	+ pragmas: Option<Vec<String>>,
	663	+) -> DatabaseConfig {
	664	+ DatabaseConfig {
	665	+ url: url.to_string(),
	666	+ max_connections,
	667	+ kind: Some(kind),
	668	+ init_sql: None,
	669	+ pragmas,
	670	+ acquire_timeout: Some(std::time::Duration::from_secs(30)),
	671	+ idle_timeout: None,
	672	+ max_lifetime: Some(std::time::Duration::from_secs(1800)),
	673	+ health_check_interval: None,
	674	+ default_isolation: None,
	675	+ }
	676	+}
	677	+
368	678	/// Manager for multiple named database pools.
369	679	#[derive(Default)]
370	680	pub struct PoolManager {
@@ -382,14 +692,7 @@ impl PoolManager {
382	692	config: DatabaseConfig,
383	693	) -> Result<AnyPool, sqlx::Error> {
384	694	let name = name.into();
385		- let kind = config.kind.unwrap_or_else(\|\| DbKind::from_url(&config.url));
386		- let pool = AnyPool::connect_with(
387		- &config.url,
388		- config.max_connections,
389		- kind,
390		- config.pragmas.clone(),
391		- )
392		- .await?;
	695	+ let pool = AnyPool::connect(config).await?;
393	696	self.pools.write().await.insert(name, pool.clone());
394	697	Ok(pool)
395	698	}
@@ -414,6 +717,25 @@ impl PoolManager {
414	717	(healthy, total)
415	718	}
416	719
	720	+ /// Spawn a background health-check loop for all pools. Pings every pool at
	721	+ /// `interval` and logs unhealthy ones. Errors are logged, never propagated —
	722	+ /// a flaky health check must not take the server down.
	723	+ pub async fn spawn_health_checks(self: &Arc<Self>, interval: Duration) {
	724	+ let pools = self.pools.clone();
	725	+ tokio::spawn(async move {
	726	+ let mut ticker = tokio::time::interval(interval);
	727	+ loop {
	728	+ ticker.tick().await;
	729	+ let snapshot = pools.read().await;
	730	+ for (name, pool) in snapshot.iter() {
	731	+ if let Err(e) = pool.health_check().await {
	732	+ tracing::warn!(pool = %name, error = %e, "database health check failed");
	733	+ }
	734	+ }
	735	+ }
	736	+ });
	737	+ }
	738	+
417	739	pub async fn len(&self) -> usize {
418	740	self.pools.read().await.len()
419	741	}
@@ -432,7 +754,9 @@ mod tests {
432	754	assert_eq!(DbKind::from_url("postgres://localhost/db"), DbKind::Postgres);
433	755	assert_eq!(DbKind::from_url("postgresql://localhost/db"), DbKind::Postgres);
434	756	assert_eq!(DbKind::from_url("sqlite://./test.db"), DbKind::Sqlite);
	757	+ assert_eq!(DbKind::from_url("sqlite::memory:"), DbKind::Sqlite);
435	758	assert_eq!(DbKind::from_url("mysql://localhost/db"), DbKind::MySql);
	759	+ assert_eq!(DbKind::from_url("mariadb://localhost/db"), DbKind::MySql);
436	760	}
437	761
438	762	#[test]
@@ -441,5 +765,75 @@ mod tests {
441	765	assert_eq!(cfg.max_connections, 10);
442	766	assert!(cfg.url.is_empty());
443	767	assert!(cfg.kind.is_none());
	768	+ assert!(cfg.acquire_timeout.is_some());
	769	+ assert!(cfg.health_check_interval.is_none());
	770	+ }
	771	+
	772	+ #[test]
	773	+ fn test_isolation_support() {
	774	+ assert!(IsolationLevel::Serializable.is_supported(DbKind::Sqlite));
	775	+ assert!(!IsolationLevel::ReadCommitted.is_supported(DbKind::Sqlite));
	776	+ assert!(IsolationLevel::ReadCommitted.is_supported(DbKind::Postgres));
	777	+ }
	778	+
	779	+ #[tokio::test]
	780	+ async fn test_sqlite_roundtrip_types() {
	781	+ let pool =
	782	+ AnyPool::connect(config("sqlite::memory:", 1, DbKind::Sqlite, None)).await.unwrap();
	783	+ pool.execute(
	784	+ "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT, ok INTEGER, data BLOB, big REAL)",
	785	+ )
	786	+ .await
	787	+ .unwrap();
	788	+ pool.execute_with(
	789	+ "INSERT INTO t (name, ok, data, big) VALUES (?, ?, ?, ?)",
	790	+ &[
	791	+ Param::Text("hello".into()),
	792	+ Param::Int(1),
	793	+ Param::Bytes(vec![1, 2, 3]),
	794	+ Param::Float(f64::INFINITY),
	795	+ ],
	796	+ )
	797	+ .await
	798	+ .unwrap();
	799	+ let rows = pool.query_with("SELECT * FROM t", &[]).await.unwrap();
	800	+ let row = rows.as_array().unwrap()[0].as_object().unwrap();
	801	+ assert_eq!(row["name"], serde_json::json!("hello"));
	802	+ assert_eq!(row["ok"], serde_json::json!(1));
	803	+ // BLOB → base64 object.
	804	+ assert_eq!(row["data"], serde_json::json!({ "$bytes": B64.encode([1u8, 2, 3]) }));
	805	+ // Non-finite float → null (JSON-safe).
	806	+ assert_eq!(row["big"], serde_json::Value::Null);
	807	+ }
	808	+
	809	+ #[tokio::test]
	810	+ async fn test_transaction_detects_writes_and_reads() {
	811	+ let pool =
	812	+ AnyPool::connect(config("sqlite::memory:", 1, DbKind::Sqlite, None)).await.unwrap();
	813	+ pool.execute("CREATE TABLE u (id INTEGER PRIMARY KEY, n INTEGER)").await.unwrap();
	814	+ // A write followed by a read inside one transaction.
	815	+ let res = pool
	816	+ .transaction(&[
	817	+ ("INSERT INTO u (n) VALUES (?)".into(), vec![serde_json::json!(5)]),
	818	+ ("SELECT SUM(n) AS total FROM u".into(), vec![]),
	819	+ ])
	820	+ .await
	821	+ .unwrap();
	822	+ assert_eq!(res, serde_json::json!([{ "total": 5 }]));
	823	+ }
	824	+
	825	+ #[tokio::test]
	826	+ async fn test_stream_chunks() {
	827	+ let pool =
	828	+ AnyPool::connect(config("sqlite::memory:", 1, DbKind::Sqlite, None)).await.unwrap();
	829	+ pool.execute("CREATE TABLE s (v INTEGER)").await.unwrap();
	830	+ for i in 0..5 {
	831	+ pool.execute_with("INSERT INTO s (v) VALUES (?)", &[Param::Int(i)]).await.unwrap();
	832	+ }
	833	+ let chunks = pool.query_stream("SELECT v FROM s ORDER BY v", &[], 2).await.unwrap();
	834	+ // 5 rows in chunks of 2 → [2,2,1].
	835	+ assert_eq!(chunks.len(), 3);
	836	+ assert_eq!(chunks[0].as_array().unwrap().len(), 2);
	837	+ assert_eq!(chunks[2].as_array().unwrap().len(), 1);
444	838	}
445	839	}
446	840

▼crates/justapi-core/tests/db_bridge.rs

+170-0

@@ -0,0 +1,170 @@
	1	+//! Integration tests for the database bridge: type fidelity, blob binding,
	2	+//! robust transaction detection, streaming, isolation, and multi-engine URL
	3	+//! routing. SQLite runs in-memory (always available); Postgres/MySQL tests are
	4	+//! skipped unless a `JUSTAPI_TEST_PG_URL` / `JUSTAPI_TEST_MYSQL_URL` env var is
	5	+//! provided, so the suite stays hermetic by default.
	6	+
	7	+use base64::Engine;
	8	+use justapi_core::db::{AnyPool, DatabaseConfig, DbKind, IsolationLevel, Param};
	9	+
	10	+async fn sqlite_pool() -> AnyPool {
	11	+ AnyPool::connect(DatabaseConfig {
	12	+ url: "sqlite::memory:".into(),
	13	+ max_connections: 1,
	14	+ kind: Some(DbKind::Sqlite),
	15	+ ..Default::default()
	16	+ })
	17	+ .await
	18	+ .unwrap()
	19	+}
	20	+
	21	+#[tokio::test]
	22	+async fn sqlite_types_and_blobs() {
	23	+ let pool = sqlite_pool().await;
	24	+ pool.execute(
	25	+ "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT, ok INTEGER, data BLOB, big REAL, nul TEXT)",
	26	+ )
	27	+ .await
	28	+ .unwrap();
	29	+ pool.execute_with(
	30	+ "INSERT INTO t (name, ok, data, big, nul) VALUES (?, ?, ?, ?, ?)",
	31	+ &[
	32	+ Param::Text("hi".into()),
	33	+ Param::Int(1),
	34	+ Param::Bytes(vec![0xDE, 0xAD, 0xBE, 0xEF]),
	35	+ Param::Float(f64::NAN),
	36	+ Param::Null,
	37	+ ],
	38	+ )
	39	+ .await
	40	+ .unwrap();
	41	+ let rows = pool.query_with("SELECT * FROM t", &[]).await.unwrap();
	42	+ let row = rows.as_array().unwrap()[0].as_object().unwrap();
	43	+ assert_eq!(row["name"], serde_json::json!("hi"));
	44	+ assert_eq!(row["ok"], serde_json::json!(1));
	45	+ assert_eq!(row["nul"], serde_json::Value::Null);
	46	+ assert_eq!(
	47	+ row["data"],
	48	+ serde_json::json!({ "$bytes": base64::engine::general_purpose::STANDARD.encode([0xDE, 0xAD, 0xBE, 0xEF]) })
	49	+ );
	50	+ assert_eq!(row["big"], serde_json::Value::Null);
	51	+}
	52	+
	53	+#[tokio::test]
	54	+async fn sqlite_transaction_read_and_write() {
	55	+ let pool = sqlite_pool().await;
	56	+ pool.execute("CREATE TABLE u (id INTEGER PRIMARY KEY, n INTEGER)").await.unwrap();
	57	+ let res = pool
	58	+ .transaction(&[
	59	+ ("INSERT INTO u (n) VALUES (?)".into(), vec![serde_json::json!(5)]),
	60	+ ("INSERT INTO u (n) VALUES (?)".into(), vec![serde_json::json!(7)]),
	61	+ ("SELECT SUM(n) AS total FROM u".into(), vec![]),
	62	+ ])
	63	+ .await
	64	+ .unwrap();
	65	+ assert_eq!(res, serde_json::json!([{ "total": 12 }]));
	66	+}
	67	+
	68	+#[tokio::test]
	69	+async fn sqlite_transaction_isolation_serializable() {
	70	+ let pool = sqlite_pool().await;
	71	+ pool.execute("CREATE TABLE v (n INTEGER)").await.unwrap();
	72	+ let res = pool
	73	+ .transaction_with_isolation(
	74	+ &[("INSERT INTO v (n) VALUES (?)".into(), vec![Param::Int(1)])],
	75	+ Some(IsolationLevel::Serializable),
	76	+ )
	77	+ .await
	78	+ .unwrap();
	79	+ assert_eq!(res, serde_json::json!({ "rows_affected": 1 }));
	80	+}
	81	+
	82	+#[tokio::test]
	83	+async fn sqlite_stream_chunks() {
	84	+ let pool = sqlite_pool().await;
	85	+ pool.execute("CREATE TABLE s (v INTEGER)").await.unwrap();
	86	+ for i in 0..7 {
	87	+ pool.execute_with("INSERT INTO s (v) VALUES (?)", &[Param::Int(i)]).await.unwrap();
	88	+ }
	89	+ let chunks = pool.query_stream("SELECT v FROM s ORDER BY v", &[], 3).await.unwrap();
	90	+ assert_eq!(chunks.len(), 3);
	91	+ assert_eq!(chunks[0].as_array().unwrap().len(), 3);
	92	+ assert_eq!(chunks[2].as_array().unwrap().len(), 1);
	93	+}
	94	+
	95	+#[tokio::test]
	96	+async fn sqlite_insert_returning() {
	97	+ let pool = sqlite_pool().await;
	98	+ pool.execute("CREATE TABLE r (id INTEGER PRIMARY KEY, name TEXT)").await.unwrap();
	99	+ let row = pool
	100	+ .insert_returning("r", &["name".into()], &serde_json::json!({ "name": "x" }))
	101	+ .await
	102	+ .unwrap();
	103	+ assert_eq!(row.as_array().unwrap()[0]["name"], serde_json::json!("x"));
	104	+}
	105	+
	106	+#[tokio::test]
	107	+async fn db_kind_routing() {
	108	+ assert_eq!(DbKind::from_url("postgres://h/db"), DbKind::Postgres);
	109	+ assert_eq!(DbKind::from_url("postgresql://h/db"), DbKind::Postgres);
	110	+ assert_eq!(DbKind::from_url("sqlite::memory:"), DbKind::Sqlite);
	111	+ assert_eq!(DbKind::from_url("sqlite://./x"), DbKind::Sqlite);
	112	+ assert_eq!(DbKind::from_url("mysql://h/db"), DbKind::MySql);
	113	+ assert_eq!(DbKind::from_url("mariadb://h/db"), DbKind::MySql);
	114	+}
	115	+
	116	+/// Postgres round-trip (skipped unless JUSTAPI_TEST_PG_URL is set).
	117	+#[tokio::test]
	118	+async fn pg_roundtrip_if_available() {
	119	+ let url = match std::env::var("JUSTAPI_TEST_PG_URL") {
	120	+ Ok(u) => u,
	121	+ Err(_) => return,
	122	+ };
	123	+ let pool = AnyPool::connect(DatabaseConfig {
	124	+ url,
	125	+ max_connections: 4,
	126	+ kind: Some(DbKind::Postgres),
	127	+ init_sql: Some(
	128	+ "DROP TABLE IF EXISTS pg_demo; CREATE TABLE pg_demo (id SERIAL PRIMARY KEY, name TEXT, qty INT)".into(),
	129	+ ),
	130	+ ..Default::default()
	131	+ })
	132	+ .await
	133	+ .unwrap();
	134	+ pool.execute_with(
	135	+ "INSERT INTO pg_demo (name, qty) VALUES (?, ?)",
	136	+ &[Param::Text("a".into()), Param::Int(3)],
	137	+ )
	138	+ .await
	139	+ .unwrap();
	140	+ let rows = pool.query_with("SELECT * FROM pg_demo ORDER BY id", &[]).await.unwrap();
	141	+ assert_eq!(rows.as_array().unwrap()[0]["name"], serde_json::json!("a"));
	142	+}
	143	+
	144	+/// MySQL round-trip (skipped unless JUSTAPI_TEST_MYSQL_URL is set).
	145	+#[tokio::test]
	146	+async fn mysql_roundtrip_if_available() {
	147	+ let url = match std::env::var("JUSTAPI_TEST_MYSQL_URL") {
	148	+ Ok(u) => u,
	149	+ Err(_) => return,
	150	+ };
	151	+ let pool = AnyPool::connect(DatabaseConfig {
	152	+ url,
	153	+ max_connections: 4,
	154	+ kind: Some(DbKind::MySql),
	155	+ init_sql: Some(
	156	+ "DROP TABLE IF EXISTS my_demo; CREATE TABLE my_demo (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64), qty INT)".into(),
	157	+ ),
	158	+ ..Default::default()
	159	+ })
	160	+ .await
	161	+ .unwrap();
	162	+ pool.execute_with(
	163	+ "INSERT INTO my_demo (name, qty) VALUES (?, ?)",
	164	+ &[Param::Text("a".into()), Param::Int(3)],
	165	+ )
	166	+ .await
	167	+ .unwrap();
	168	+ let rows = pool.query_with("SELECT * FROM my_demo ORDER BY id", &[]).await.unwrap();
	169	+ assert_eq!(rows.as_array().unwrap()[0]["name"], serde_json::json!("a"));
	170	+}
0	171

▼crates/justapi-py/Cargo.toml

+1-0

@@ -18,6 +18,7 @@ http = "1"
18	18	futures = "0.3"
19	19	serde.workspace = true
20	20	serde_json.workspace = true
	21	+base64 = { workspace = true }
21	22	serde_urlencoded.workspace = true
22	23	tracing.workspace = true
23	24	anyhow.workspace = true
24	25

▼crates/justapi-py/python/justapi/__init__.py

+2-2

@@ -30,7 +30,7 @@ from . import tracing as tracing
30	30	from . import status as status
31	31
32	32	from ._justapi import serve, TokenStreamResponse, ValidatedStreamResponse, Dag, DagNode, RateLimiter, RateLimitResult, Request, HTTPConnection, Headers, QueryParams, URL, State, RequestStream, UploadFile # type: ignore[import-untyped]
33		-from ._justapi import Database, DbPool # type: ignore[import-untyped]
	33	+from ._justapi import Database, DbPool, DbParam # type: ignore[import-untyped]
34	34	from .app import JustAPIApp, JustAPI, Depends, adaptive_batch, APIRouter, Controller, controller, route_get, route_post, route_put, route_patch, route_delete, route_query, route_sse, route_websocket, JustAPITestClient, RequestValidationError, Session
35	35	from .system import build_help, build_openapi, register_system_routes
36	36	from .exceptions import HTTPException, WebSocketException
@@ -167,7 +167,7 @@ def pydantic_schema(model_class) -> str:
167	167
168	168
169	169	__all__ = [
170		- "serve", "JustAPIApp", "JustAPI", "Depends", "Database", "DbPool", "Schema", "pydantic_schema",
	170	+ "serve", "JustAPIApp", "JustAPI", "Depends", "Database", "DbPool", "DbParam", "Schema", "pydantic_schema",
171	171	"JustAPITestClient", "testing", "tracing", "Jinja2Templates", "BackgroundTasks",
172	172	"TokenStreamResponse", "ValidatedStreamResponse", "WebSocket", "Dag", "DagNode", "RateLimiter", "RateLimitResult",
173	173	"adaptive_batch", "APIRouter", "Controller", "controller",
174	174

▼crates/justapi-py/python/justapi/app.py

+45-16

@@ -866,24 +866,48 @@ class JustAPIApp:
866	866	def use(self, plugin):
867	867	self._app.use(plugin)
868	868
869		- def set_database(self, db, init_sql=None, pragmas=None, wal=False):
	869	+ def set_database(
	870	+ self,
	871	+ db,
	872	+ init_sql=None,
	873	+ pragmas=None,
	874	+ wal=False,
	875	+ acquire_timeout=None,
	876	+ idle_timeout=None,
	877	+ max_lifetime=None,
	878	+ health_check_interval=None,
	879	+ isolation=None,
	880	+ ):
	881	+ def _build(existing):
	882	+ return Database(
	883	+ existing.url,
	884	+ max_connections=existing.max_connections,
	885	+ init_sql=init_sql if init_sql is not None else existing.init_sql,
	886	+ pragmas=pragmas if pragmas is not None else existing.pragmas,
	887	+ acquire_timeout=acquire_timeout
	888	+ if acquire_timeout is not None
	889	+ else existing.acquire_timeout,
	890	+ idle_timeout=idle_timeout if idle_timeout is not None else existing.idle_timeout,
	891	+ max_lifetime=max_lifetime if max_lifetime is not None else existing.max_lifetime,
	892	+ health_check_interval=health_check_interval
	893	+ if health_check_interval is not None
	894	+ else existing.health_check_interval,
	895	+ isolation=isolation if isolation is not None else existing.isolation,
	896	+ )
	897	+
870	898	if isinstance(db, str):
871		- db = Database(db, init_sql=init_sql, pragmas=pragmas)
	899	+ db = Database(
	900	+ db,
	901	+ init_sql=init_sql,
	902	+ pragmas=pragmas,
	903	+ acquire_timeout=acquire_timeout,
	904	+ idle_timeout=idle_timeout,
	905	+ max_lifetime=max_lifetime,
	906	+ health_check_interval=health_check_interval,
	907	+ isolation=isolation,
	908	+ )
872	909	else:
873		- if init_sql is not None:
874		- db = Database(
875		- db.url,
876		- max_connections=db.max_connections,
877		- init_sql=init_sql,
878		- pragmas=pragmas if pragmas is not None else db.pragmas,
879		- )
880		- elif pragmas is not None:
881		- db = Database(
882		- db.url,
883		- max_connections=db.max_connections,
884		- init_sql=db.init_sql,
885		- pragmas=pragmas,
886		- )
	910	+ db = _build(db)
887	911	if wal:
888	912	existing = list(db.pragmas or [])
889	913	if "journal_mode=WAL" not in existing:
@@ -893,6 +917,11 @@ class JustAPIApp:
893	917	max_connections=db.max_connections,
894	918	init_sql=db.init_sql,
895	919	pragmas=existing,
	920	+ acquire_timeout=db.acquire_timeout,
	921	+ idle_timeout=db.idle_timeout,
	922	+ max_lifetime=db.max_lifetime,
	923	+ health_check_interval=db.health_check_interval,
	924	+ isolation=db.isolation,
896	925	)
897	926	self._app.set_database(db)
898	927
899	928

▼crates/justapi-py/src/database.rs

+161-6

@@ -1,7 +1,7 @@
1	1	use pyo3::prelude::*;
2	2	use pyo3::types::{PyAnyMethods, PyDict, PyList};
3	3
4		-use justapi_core::db::DatabaseConfig;
	4	+use justapi_core::db::{DatabaseConfig, IsolationLevel};
5	5
6	6	/// A database connection pool configuration for JustAPI.
7	7	///
@@ -19,19 +19,50 @@ pub struct Database {
19	19	/// `journal_mode=WAL`). Ignored by Postgres/MySQL.
20	20	#[pyo3(get)]
21	21	pub pragmas: Option<Vec<String>>,
	22	+ /// Max seconds to wait for a connection from the pool.
	23	+ #[pyo3(get)]
	24	+ pub acquire_timeout: Option<f64>,
	25	+ /// Max seconds a connection may stay idle before recycling.
	26	+ #[pyo3(get)]
	27	+ pub idle_timeout: Option<f64>,
	28	+ /// Max seconds a connection may live before being closed.
	29	+ #[pyo3(get)]
	30	+ pub max_lifetime: Option<f64>,
	31	+ /// Seconds between background health-check pings (0/None disables).
	32	+ #[pyo3(get)]
	33	+ pub health_check_interval: Option<f64>,
	34	+ /// Default transaction isolation level: "read-uncommitted",
	35	+ /// "read-committed", "repeatable-read", "serializable", "snapshot".
	36	+ #[pyo3(get)]
	37	+ pub isolation: Option<String>,
22	38	}
23	39
24	40	#[pymethods]
25	41	impl Database {
26	42	#[new]
27		- #[pyo3(signature = (url, max_connections=10, init_sql=None, pragmas=None))]
	43	+ #[pyo3(signature = (url, max_connections=10, init_sql=None, pragmas=None, acquire_timeout=None, idle_timeout=None, max_lifetime=None, health_check_interval=None, isolation=None))]
28	44	fn py_new(
29	45	url: String,
30	46	max_connections: u32,
31	47	init_sql: Option<String>,
32	48	pragmas: Option<Vec<String>>,
	49	+ acquire_timeout: Option<f64>,
	50	+ idle_timeout: Option<f64>,
	51	+ max_lifetime: Option<f64>,
	52	+ health_check_interval: Option<f64>,
	53	+ isolation: Option<String>,
33	54	) -> Self {
34		- Self { url, max_connections, init_sql, pragmas }
	55	+ Self {
	56	+ url,
	57	+ max_connections,
	58	+ init_sql,
	59	+ pragmas,
	60	+ acquire_timeout,
	61	+ idle_timeout,
	62	+ max_lifetime,
	63	+ health_check_interval,
	64	+ isolation,
	65	+ }
35	66	}
36	67
37	68	fn __repr__(&self) -> String {
@@ -41,20 +72,83 @@ impl Database {
41	72
42	73	impl Database {
43	74	pub fn new(url: String, max_connections: u32) -> Self {
44		- Self { url, max_connections, init_sql: None, pragmas: None }
	75	+ Self {
	76	+ url,
	77	+ max_connections,
	78	+ init_sql: None,
	79	+ pragmas: None,
	80	+ acquire_timeout: Some(30.0),
	81	+ idle_timeout: None,
	82	+ max_lifetime: Some(1800.0),
	83	+ health_check_interval: None,
	84	+ isolation: None,
	85	+ }
45	86	}
46	87
47	88	pub fn to_config(&self) -> DatabaseConfig {
	89	+ let isolation = self.isolation.as_deref().and_then(parse_isolation);
48	90	DatabaseConfig {
49	91	url: self.url.clone(),
50	92	max_connections: self.max_connections,
51	93	kind: None,
52	94	init_sql: self.init_sql.clone(),
53	95	pragmas: self.pragmas.clone(),
	96	+ acquire_timeout: self.acquire_timeout.map(std::time::Duration::from_secs_f64),
	97	+ idle_timeout: self.idle_timeout.map(std::time::Duration::from_secs_f64),
	98	+ max_lifetime: self.max_lifetime.map(std::time::Duration::from_secs_f64),
	99	+ health_check_interval: self
	100	+ .health_check_interval
	101	+ .map(std::time::Duration::from_secs_f64),
	102	+ default_isolation: isolation,
54	103	}
55	104	}
56	105	}
57	106
	107	+/// Parse a Python isolation-level string into the Rust enum.
	108	+fn parse_isolation(s: &str) -> Option<IsolationLevel> {
	109	+ match s.to_ascii_lowercase().replace(['-', '_'], "").as_str() {
	110	+ "readuncommitted" => Some(IsolationLevel::ReadUncommitted),
	111	+ "readcommitted" => Some(IsolationLevel::ReadCommitted),
	112	+ "repeatableread" => Some(IsolationLevel::RepeatableRead),
	113	+ "serializable" => Some(IsolationLevel::Serializable),
	114	+ "snapshot" => Some(IsolationLevel::Snapshot),
	115	+ _ => None,
	116	+ }
	117	+}
	118	+
	119	+/// A typed query parameter for the `DbPool` bridge.
	120	+///
	121	+/// Most values pass through as JSON, but some need an explicit type that JSON
	122	+/// cannot represent losslessly — chiefly `bytes` (BLOBs). Use `DbParam.bytes(...)`
	123	+/// for binary columns. The marker serializes to `{"$bytes": "<base64>"}` and is
	124	+/// rebound as a real BLOB on the Rust side (round-tripping back as
	125	+/// `{"$bytes": "<base64>"}`).
	126	+#[pyclass(name = "DbParam", from_py_object)]
	127	+#[derive(Clone)]
	128	+pub struct DbParam {
	129	+ wire: serde_json::Value,
	130	+}
	131	+
	132	+#[pymethods]
	133	+impl DbParam {
	134	+ /// A BLOB/binary parameter. `data` is `bytes`.
	135	+ #[staticmethod]
	136	+ fn bytes(data: &[u8]) -> Self {
	137	+ use base64::engine::general_purpose::STANDARD as B64;
	138	+ use base64::Engine;
	139	+ let mut m = serde_json::Map::new();
	140	+ m.insert("$bytes".into(), serde_json::Value::String(B64.encode(data)));
	141	+ Self { wire: serde_json::Value::Object(m) }
	142	+ }
	143	+}
	144	+
	145	+impl DbParam {
	146	+ /// The JSON wire marker the Rust side understands.
	147	+ pub fn wire(&self) -> &serde_json::Value {
	148	+ &self.wire
	149	+ }
	150	+}
	151	+
58	152	/// Python-facing handle to a resolved `AnyPool`.
59	153	///
60	154	/// Every method runs the SQL in Rust (over `sqlx::Any`) with **bound
@@ -140,11 +234,15 @@ impl DbPool {
140	234	/// Run many `(sql, params)` statements atomically in one transaction and
141	235	/// commit. `stmts` is a list of `[sql, params]` pairs (params optional).
142	236	/// Returns the rows of the final statement if it was a query, else
143		- /// `{"rows_affected": N}`.
	237	+ /// `{"rows_affected": N}`. `isolation` optionally sets the transaction
	238	+ /// isolation level (e.g. `"serializable"`); unsupported levels fall back to
	239	+ /// the engine default.
	240	+ #[pyo3(signature = (stmts, isolation=None))]
144	241	fn transaction<'py>(
145	242	&self,
146	243	py: Python<'py>,
147	244	stmts: Vec<(String, Option<Vec<Py<PyAny>>>)>,
	245	+ isolation: Option<String>,
148	246	) -> PyResult<Bound<'py, PyAny>> {
149	247	let stmts_json: Vec<(String, Vec<serde_json::Value>)> = stmts
150	248	.into_iter()
@@ -159,18 +257,73 @@ impl DbPool {
159	257	Ok::<_, pyo3::PyErr>((sql, ps))
160	258	})
161	259	.collect::<PyResult<Vec<_>>>()?;
	260	+ let iso = isolation.as_deref().and_then(parse_isolation);
162	261	let inner = self.inner.clone();
163	262	let rt = self.rt.clone();
164	263	let res: serde_json::Value = py
165	264	.detach(move \|\| {
166	265	rt.block_on(async move {
167		- inner.transaction(&stmts_json).await.map_err(\|e\| anyhow::anyhow!(e.to_string()))
	266	+ match iso {
	267	+ Some(level) => {
	268	+ let typed: Vec<(String, Vec<_>)> = stmts_json
	269	+ .iter()
	270	+ .map(\|(s, p)\| {
	271	+ (
	272	+ s.clone(),
	273	+ p.iter().map(justapi_core::db::Param::from).collect(),
	274	+ )
	275	+ })
	276	+ .collect();
	277	+ inner
	278	+ .transaction_with_isolation(&typed, Some(level))
	279	+ .await
	280	+ .map_err(\|e\| anyhow::anyhow!(e.to_string()))
	281	+ }
	282	+ None => inner
	283	+ .transaction(&stmts_json)
	284	+ .await
	285	+ .map_err(\|e\| anyhow::anyhow!(e.to_string())),
	286	+ }
168	287	})
169	288	})
170	289	.map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
171	290	json_to_python(py, &res)
172	291	}
173	292
	293	+ /// Stream a query in batches of `chunk` rows, returning a list of row-chunks
	294	+ /// (each chunk is itself a list of row dicts). Keeps large result sets
	295	+ /// bounded in memory on the Rust side. `params` optional.
	296	+ #[pyo3(signature = (sql, params=None, chunk=1000))]
	297	+ fn query_stream<'py>(
	298	+ &self,
	299	+ py: Python<'py>,
	300	+ sql: String,
	301	+ params: Option<Vec<Py<PyAny>>>,
	302	+ chunk: usize,
	303	+ ) -> PyResult<Bound<'py, PyAny>> {
	304	+ let params_json: Vec<serde_json::Value> = match params {
	305	+ Some(ps) => {
	306	+ ps.iter().map(\|p\| python_to_json(py, p.bind(py))).collect::<PyResult<Vec<_>>>()?
	307	+ }
	308	+ None => Vec::new(),
	309	+ };
	310	+ let inner = self.inner.clone();
	311	+ let rt = self.rt.clone();
	312	+ let chunks: Vec<serde_json::Value> = py
	313	+ .detach(move \|\| {
	314	+ rt.block_on(async move {
	315	+ let typed: Vec<_> =
	316	+ params_json.iter().map(justapi_core::db::Param::from).collect();
	317	+ inner
	318	+ .query_stream(&sql, &typed, chunk)
	319	+ .await
	320	+ .map_err(\|e\| anyhow::anyhow!(e.to_string()))
	321	+ })
	322	+ })
	323	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
	324	+ json_to_python(py, &serde_json::Value::Array(chunks))
	325	+ }
	326	+
174	327	/// Insert a row from a dict and return the inserted row (via `RETURNING *`).
175	328	/// `columns` restricts which keys are written (injection guard).
176	329	#[pyo3(signature = (table, data, columns=None))]
@@ -274,6 +427,8 @@ fn python_to_json(py: Python<'_>, obj: &Bound<'_, PyAny>) -> PyResult<serde_json
274	427	out[key] = python_to_json(py, &v)?;
275	428	}
276	429	Ok(out)
	430	+ } else if let Ok(p) = obj.extract::<DbParam>() {
	431	+ Ok(p.wire().clone())
277	432	} else {
278	433	Ok(serde_json::Value::String(obj.str()?.to_string()))
279	434	}
280	435

▼crates/justapi-py/src/lib.rs

+2-1

@@ -14,7 +14,7 @@ mod test_client;
14	14	mod websocket;
15	15
16	16	pub use dag::{Dag, DagNode};
17		-pub use database::{Database, DbPool};
	17	+pub use database::{Database, DbParam, DbPool};
18	18	pub use multipart::UploadFile;
19	19	pub use native::{validate_value, TokenStreamResponse, ValidatedStreamResponse};
20	20	pub use rate_limit::PyRateLimitResult;
@@ -69,6 +69,7 @@ fn _justapi(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
69	69	m.add_class::<UploadFile>()?;
70	70	m.add_class::<Database>()?;
71	71	m.add_class::<DbPool>()?;
	72	+ m.add_class::<DbParam>()?;
72	73	m.add_class::<Dag>()?;
73	74	m.add_class::<DagNode>()?;
74	75	m.add_class::<background::BackgroundTasks>()?;
75	76

▼crates/justapi-py/src/native/app.rs

+5-0

@@ -1437,6 +1437,11 @@ impl JustAPIApp {
1437	1437	}
1438	1438	}
1439	1439	}
	1440	+ // Optional background health-check loop.
	1441	+ if let Some(interval) = config.health_check_interval {
	1442	+ let mgr = std::sync::Arc::new(mgr);
	1443	+ mgr.spawn_health_checks(interval).await;
	1444	+ }
1440	1445	Some(pool)
1441	1446	}
1442	1447	Err(e) => {
1443	1448

▼crates/justapi-py/src/test_client.rs

+1-0

@@ -91,6 +91,7 @@ impl JustAPITestClient {
91	91	max_connections: 5,
92	92	init_sql: None,
93	93	pragmas: None,
	94	+ ..Default::default()
94	95	};
95	96	let pool = py
96	97	.detach(move \|\| -> Result<AnyPool, String> {
97	98