swadhin / justapi public

swadhinbiswas

[email protected]
2026-07-24 18:46:22 +0600 committed NaN years ago
f7229e3
commit f7229e3
15 files changed
+358 -241
Showing 15 changed files+358-241
Dockerfilebenchmarks/smoke_dbpool.pybenchmarks/smoke_dbpool_prerun.pybenchmarks/workloads_crud_pg.pycrates/justapi-core/src/extract.rscrates/justapi-core/src/memory.rscrates/justapi-core/src/router.rscrates/justapi-core/src/secrets.rscrates/justapi-core/src/static_files.rscrates/justapi-core/src/wasm.rscrates/justapi-py/python/justapi/_native_helper.pycrates/justapi-py/python/justapi/app.pycrates/justapi-py/src/gil_pool.rscrates/justapi-py/src/native/app.rscrates/justapi-py/src/native/handlers.rs
▼Dockerfile
+7-1
@@ -1,6 +1,7 @@
1
1
# =============================================================================
2
2
# Stage 1: Build the Rust CLI binary and Python wheel
3
3
# =============================================================================
4
+# Pin base image digest for reproducible builds. Run `docker pull rust:1.75-bookworm && docker inspect --format='{{index .RepoDigests 0}}' rust:1.75-bookworm` to get the current digest.
4
5
FROM rust:1.75-bookworm AS builder
5
6
6
7
# Install dependencies including python for building maturin wheels
@@ -18,7 +19,7 @@ COPY python/ python/
18
19
RUN python3 -m venv /opt/venv
19
20
ENV PATH="/opt/venv/bin:$PATH"
20
21
21
-RUN pip install maturin
22
+RUN pip install maturin==1.8.3
22
23
23
24
# Build the Rust CLI
24
25
RUN cargo build --release -p justapi-cli --features tls,compression
@@ -30,6 +31,7 @@ RUN maturin build -m crates/justapi-py/Cargo.toml --release -o wheels/
30
31
# =============================================================================
31
32
# Stage 2: Minimal runtime image with Python
32
33
# =============================================================================
34
+# Pin base image digest for reproducible builds.
33
35
FROM python:3.12-slim-bookworm
34
36
35
37
RUN apt-get update && apt-get install -y --no-install-recommends \
@@ -49,4 +51,8 @@ COPY demo_app.py /app/
49
51
50
52
EXPOSE 8080
51
53
54
+# Security: drop root privileges
55
+RUN useradd -m -u 1000 -s /bin/bash justapi
56
+USER justapi
57
+
52
58
ENTRYPOINT ["python", "demo_app.py"]
53
59
▼benchmarks/smoke_dbpool.py
+2-4
@@ -6,10 +6,8 @@ import pytest
6
6
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "crates", "justapi-py"))
7
7
from justapi import JustAPIApp, Database # noqa: E402
8
8
9
-PG_URL = (
10
- "postgres://avnadmin:<SECRET_PASSWORD>"
11
- "@dexmorgan-examhallorhell-5f8c.d.aivencloud.com:11100/defaultdb?sslmode=require"
12
-)
9
+# Set JUSTAPI_PG_URL env var for DB benchmarks — never commit live credentials
10
+PG_URL = os.environ.get("JUSTAPI_PG_URL", "")
13
11
14
12
import threading
15
13
16
14
▼benchmarks/smoke_dbpool_prerun.py
+2-1
@@ -1,7 +1,8 @@
1
1
import time, threading, urllib.request
2
2
from justapi import JustAPIApp, Database
3
3
4
-URL = "postgres://avnadmin:<SECRET_PASSWORD>@dexmorgan-examhallorhell-5f8c.d.aivencloud.com:11100/defaultdb?sslmode=require"
4
+# TODO: set JUSTAPI_PG_URL or pass via env var — never commit live credentials
5
+URL = None # overridden by environment
5
6
6
7
app = JustAPIApp()
7
8
app.set_database(
8
9
▼benchmarks/workloads_crud_pg.py
+3-1
@@ -3,7 +3,9 @@ import sys
3
3
from justapi import JustAPIApp, Database
4
4
5
5
PORT = sys.argv[1] if len(sys.argv) > 1 else "8099"
6
-PGURL = sys.argv[2] if len(sys.argv) > 2 else "postgres://avnadmin:<SECRET_PASSWORD>@dexmorgan-examhallorhell-5f8c.d.aivencloud.com:11100/defaultdb?sslmode=require"
6
+# Pass via CLI arg or JUSTAPI_PG_URL env var — never commit live credentials
7
+import os
8
+PGURL = sys.argv[2] if len(sys.argv) > 2 else os.environ.get("JUSTAPI_PG_URL", "")
7
9
8
10
INIT = (
9
11
"DROP TABLE IF EXISTS justapi_bench_items;"
10
12
▼crates/justapi-core/src/extract.rs
+16-11
@@ -160,21 +160,26 @@ where
160
160
// ---------------------------------------------------------------------------
161
161
162
162
/// Minimal URL-decoding (percent-encoding and '+') for cookie values.
163
+/// Processes bytes, not chars, to correctly decode multi-byte UTF-8 sequences.
163
164
fn url_decode(s: &str) -> String {
164
- let mut result = String::with_capacity(s.len());
165
- let mut chars = s.chars();
166
- while let Some(c) = chars.next() {
167
- if c == '%' {
168
- let hi = chars.next().and_then(|c| c.to_digit(16)).unwrap_or(0);
169
- let lo = chars.next().and_then(|c| c.to_digit(16)).unwrap_or(0);
170
- result.push((hi as u8 * 16 + lo as u8) as char);
171
- } else if c == '+' {
172
- result.push(' ');
165
+ let bytes = s.as_bytes();
166
+ let mut result = Vec::with_capacity(bytes.len());
167
+ let mut i = 0;
168
+ while i < bytes.len() {
169
+ if bytes[i] == b'%' && i + 2 < bytes.len() {
170
+ let hi = (bytes[i + 1] as char).to_digit(16).unwrap_or(0);
171
+ let lo = (bytes[i + 2] as char).to_digit(16).unwrap_or(0);
172
+ result.push(hi as u8 * 16 + lo as u8);
173
+ i += 3;
174
+ } else if bytes[i] == b'+' {
175
+ result.push(b' ');
176
+ i += 1;
173
177
} else {
174
- result.push(c);
178
+ result.push(bytes[i]);
179
+ i += 1;
175
180
}
176
181
}
177
- result
182
+ String::from_utf8(result).unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned())
178
183
}
179
184
180
185
// ---------------------------------------------------------------------------
181
186
▼crates/justapi-core/src/memory.rs
+4-4
@@ -80,11 +80,11 @@ impl SharedArena {
80
80
}
81
81
82
82
pub fn reset(&self) {
83
- self.inner.lock().unwrap().reset();
83
+ self.inner.lock().unwrap_or_else(|e| e.into_inner()).reset();
84
84
}
85
85
86
86
pub fn alloc_str(&self, s: &str) -> Option<String> {
87
- let mut inner = self.inner.lock().unwrap();
87
+ let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
88
88
inner.alloc_str(s).map(|r| r.to_string())
89
89
}
90
90
}
@@ -121,7 +121,7 @@ impl BufferPool {
121
121
/// Acquire a buffer with at least `min_size` capacity.
122
122
pub fn acquire(&self, min_size: usize) -> Vec<u8> {
123
123
let idx = Self::bucket_index(min_size);
124
- let mut pool = self.buckets[idx].lock().unwrap();
124
+ let mut pool = self.buckets[idx].lock().unwrap_or_else(|e| e.into_inner());
125
125
pool.pop().unwrap_or_else(|| Vec::with_capacity(BUCKET_SIZES[idx]))
126
126
}
127
127
@@ -129,7 +129,7 @@ impl BufferPool {
129
129
pub fn release(&self, mut buf: Vec<u8>) {
130
130
buf.clear();
131
131
let idx = Self::bucket_index(buf.capacity());
132
- let mut pool = self.buckets[idx].lock().unwrap();
132
+ let mut pool = self.buckets[idx].lock().unwrap_or_else(|e| e.into_inner());
133
133
if pool.len() < 128 {
134
134
pool.push(buf);
135
135
}
136
136
▼crates/justapi-core/src/router.rs
+26-21
@@ -36,45 +36,50 @@ enum CacheEntry<T> {
36
36
NotFound,
37
37
}
38
38
39
-/// Bounded, lock-free-enough route-lookup cache. A `Mutex<HashMap>` with FIFO
40
-/// eviction keeps the working set small; the critical section is a single map
41
-/// op so contention is negligible next to the I/O it saves.
39
+/// Bounded route-lookup cache. A single `Mutex<HashMap>` with embedded FIFO
40
+/// eviction avoids the deadlock-prone dual-mutex approach.
42
41
#[derive(Debug)]
43
42
struct RouteCache<T> {
44
- map: Mutex<HashMap<(Method, String), CacheEntry<T>>>,
45
- order: Mutex<Vec<(Method, String)>>,
43
+ inner: Mutex<CacheInner<T>>,
46
44
capacity: usize,
47
45
}
48
46
47
+#[derive(Debug)]
48
+struct CacheInner<T> {
49
+ map: HashMap<(Method, String), CacheEntry<T>>,
50
+ order: Vec<(Method, String)>,
51
+}
52
+
49
53
impl<T: Clone> RouteCache<T> {
50
54
fn new(capacity: usize) -> Self {
51
55
Self {
52
- map: Mutex::new(HashMap::with_capacity(capacity)),
53
- order: Mutex::new(Vec::with_capacity(capacity)),
56
+ inner: Mutex::new(CacheInner {
57
+ map: HashMap::with_capacity(capacity),
58
+ order: Vec::with_capacity(capacity),
59
+ }),
54
60
capacity: capacity.max(1),
55
61
}
56
62
}
57
63
58
- /// Look up a cached resolution. Returns `None` on a negative (not-found)
59
- /// entry or a miss.
64
+ /// Look up a cached resolution.
60
65
fn get(&self, key: &(Method, String)) -> Option<CacheEntry<T>> {
61
- self.map.lock().unwrap().get(key).cloned()
66
+ let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
67
+ inner.map.get(key).cloned()
62
68
}
63
69
64
70
/// Insert a cacheable entry, evicting the oldest key if at capacity.
65
71
fn insert(&self, key: (Method, String), entry: CacheEntry<T>) {
66
- let mut map = self.map.lock().unwrap();
67
- let mut order = self.order.lock().unwrap();
68
- if !map.contains_key(&key) {
69
- order.push(key.clone());
70
- if order.len() > self.capacity {
71
- if let Some(old) = order.first().cloned() {
72
- map.remove(&old);
73
- order.remove(0);
72
+ let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
73
+ if !inner.map.contains_key(&key) {
74
+ inner.order.push(key.clone());
75
+ if inner.order.len() > self.capacity {
76
+ if let Some(old) = inner.order.first().cloned() {
77
+ inner.map.remove(&old);
78
+ inner.order.remove(0);
74
79
}
75
80
}
76
81
}
77
- map.insert(key, entry);
82
+ inner.map.insert(key, entry);
78
83
}
79
84
}
80
85
@@ -88,13 +93,13 @@ pub struct Router<T> {
88
93
}
89
94
90
95
impl<T: Clone> Router<T> {
91
- /// Create a router with an always-on route cache (default capacity 1024).
96
+ /// Create a router with an always-on route cache (default capacity 16384).
92
97
pub fn new() -> Self {
93
98
Self {
94
99
routes: HashMap::new(),
95
100
fallback: None,
96
101
route_list: Vec::new(),
97
- cache: Some(Arc::new(RouteCache::new(1024))),
102
+ cache: Some(Arc::new(RouteCache::new(16384))),
98
103
}
99
104
}
100
105
101
106
▼crates/justapi-core/src/secrets.rs
+3-3
@@ -70,12 +70,12 @@ impl SecretsRegistry {
70
70
}
71
71
72
72
pub fn register(&self, secret: Secret) {
73
- let mut map = self.secrets.write().unwrap();
73
+ let mut map = self.secrets.write().unwrap_or_else(|e| e.into_inner());
74
74
map.insert(secret.name.clone(), secret);
75
75
}
76
76
77
77
pub fn resolve(&self, name: &str) -> Result<String, anyhow::Error> {
78
- let map = self.secrets.read().unwrap();
78
+ let map = self.secrets.read().unwrap_or_else(|e| e.into_inner());
79
79
match map.get(name) {
80
80
Some(secret) => secret.resolve(),
81
81
None => anyhow::bail!("secret '{}' not registered", name),
@@ -84,7 +84,7 @@ impl SecretsRegistry {
84
84
85
85
/// Resolve all secrets eagerly. Useful for startup validation.
86
86
pub fn resolve_all(&self) -> Vec<(String, Result<String, anyhow::Error>)> {
87
- let map = self.secrets.read().unwrap();
87
+ let map = self.secrets.read().unwrap_or_else(|e| e.into_inner());
88
88
map.iter().map(|(name, secret)| (name.clone(), secret.resolve())).collect()
89
89
}
90
90
}
91
91
▼crates/justapi-core/src/static_files.rs
+2-2
@@ -59,8 +59,8 @@ impl StaticDir {
59
59
if relative.is_empty() {
60
60
return None;
61
61
}
62
- // Block directory traversal
63
- if relative.contains("..") {
62
+ // Block directory traversal: check both literal and percent-encoded `..`
63
+ if relative.contains("..") || relative.contains("%2e") || relative.contains("%2E") {
64
64
return None;
65
65
}
66
66
let path = self.root.join(relative);
67
67
▼crates/justapi-core/src/wasm.rs
+18-4
@@ -60,10 +60,24 @@ impl WasmEngine {
60
60
let result = process_request.call_async(&mut store, (ptr, len)).await?;
61
61
62
62
let out_ptr = (result >> 32) as i32;
63
- let out_len = (result & 0xFFFFFFFF) as i32;
64
-
65
- let mut out_bytes = vec![0u8; out_len as usize];
66
- memory.read(&mut store, out_ptr as usize, &mut out_bytes)?;
63
+ let out_len_signed = (result & 0xFFFFFFFF) as i32;
64
+
65
+ if out_ptr < 0 || out_len_signed < 0 {
66
+ anyhow::bail!("WASM returned negative pointer or length");
67
+ }
68
+ let out_ptr = out_ptr as usize;
69
+ let out_len = out_len_signed as usize;
70
+
71
+ let memory_size: u64 = memory.size(&store);
72
+ let wasm_page_size: u64 = 65536;
73
+ let max_safe_len =
74
+ (memory_size * wasm_page_size).saturating_sub(out_ptr as u64).min(1_048_576) as usize;
75
+ if out_len > max_safe_len {
76
+ anyhow::bail!("WASM output length {} exceeds max safe {}", out_len, max_safe_len);
77
+ }
78
+
79
+ let mut out_bytes = vec![0u8; out_len];
80
+ memory.read(&mut store, out_ptr, &mut out_bytes)?;
67
81
68
82
let out_str = String::from_utf8(out_bytes)?;
69
83
70
84
▼crates/justapi-py/python/justapi/_native_helper.py
+5-2
@@ -85,12 +85,15 @@ def wrap_result(result):
85
85
# (BUG-1, PRODUCTION_PLAN.md P0.3).
86
86
keys = set(result.keys())
87
87
is_envelope = (
88
- "body" in result
89
- or result.get("__response__") is True
88
+ result.get("__response__") is True
90
89
or (
91
90
"status" in result
92
91
and keys <= {"status", "headers", "__response__"}
93
92
)
93
+ or (
94
+ "body" in result
95
+ and keys <= {"body", "status", "headers", "__response__"}
96
+ )
94
97
)
95
98
if is_envelope:
96
99
if "body" in result and isinstance(result["body"], str):
97
100
▼crates/justapi-py/python/justapi/app.py
+113-57
@@ -3,6 +3,7 @@ import functools
3
3
import json
4
4
import re
5
5
import typing
6
+import warnings
6
7
from urllib.parse import quote
7
8
from ._justapi import _JustAPIApp, TokenStreamResponse, ValidatedStreamResponse, Database
8
9
from .exceptions import WebSocketException
@@ -314,6 +315,8 @@ class JustAPIApp:
314
315
"""
315
316
name = kw.get("name")
316
317
if name is not None:
318
+ if name in self._named_routes and not name.startswith("builtin_"):
319
+ warnings.warn(f"Named route collision: '{name}' overwritten")
317
320
self._named_routes[name] = path
318
321
self.routes.append({
319
322
"method": method,
@@ -322,8 +325,8 @@ class JustAPIApp:
322
325
"body_schema": body_schema,
323
326
"schema": schema,
324
327
"experimental": experimental,
325
- "dependencies": [],
326
- "middlewares": [],
328
+ "dependencies": kw.get("dependencies") or [],
329
+ "middlewares": kw.get("middlewares") or [],
327
330
"tags": kw.get("tags") or [],
328
331
"summary": kw.get("summary"),
329
332
"description": kw.get("description"),
@@ -800,8 +803,8 @@ class JustAPIApp:
800
803
return responses_json, extra_json
801
804
802
805
def _route_kw(self, tags=None, summary=None, description=None, deprecated=None,
803
- status_code=None, responses=None, operation_id=None, openapi_extra=None,
804
- include_in_schema=True, name=None):
806
+ status_code=None, responses=None, operation_id=None, openapi_extra=None,
807
+ include_in_schema=True, name=None):
805
808
responses_json, extra_json = self._meta(responses, openapi_extra)
806
809
return dict(
807
810
tags=tags,
@@ -826,16 +829,16 @@ class JustAPIApp:
826
829
def _crud_noop(request):
827
830
raise AssertionError("crud routes must not invoke the Python handler")
828
831
self._app.get(path, _crud_noop, query_schema=query_schema, **kw)
829
- self._record("GET", path, _crud_noop, query_schema=query_schema, **kw)
832
+ self._record("GET", path, _crud_noop, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
830
833
return _crud_noop
831
834
if handler is None:
832
835
def decorator(func):
833
836
self._app.get(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
834
- self._record("GET", path, func, query_schema=query_schema, **kw)
837
+ self._record("GET", path, func, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
835
838
return func
836
839
return decorator
837
840
self._app.get(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
838
- self._record("GET", path, handler, query_schema=query_schema, **kw)
841
+ self._record("GET", path, handler, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
839
842
return handler
840
843
841
844
def post(self, path: str, handler=None, body_schema=None, schema=None, query_schema=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, native: bool = False, crud_table: str = None, crud_columns: typing.List[str] = None):
@@ -850,20 +853,20 @@ class JustAPIApp:
850
853
def _crud_noop(request):
851
854
raise AssertionError("crud routes must not invoke the Python handler")
852
855
self._app.post(path, _crud_noop, body_schema, schema, query_schema=query_schema, **kw)
853
- self._record("POST", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
856
+ self._record("POST", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
854
857
return _crud_noop
855
858
if handler is None:
856
859
def decorator(func):
857
860
batch_size, batch_window_ms = self._resolve_batch_config(func)
858
861
wrapped = self._wrap_batch_handler(func, native=native) if batch_size else self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
859
862
self._app.post(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
860
- self._record("POST", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
863
+ self._record("POST", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
861
864
return func
862
865
return decorator
863
866
batch_size, batch_window_ms = self._resolve_batch_config(handler)
864
867
wrapped = self._wrap_batch_handler(handler, native=native) if batch_size else self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
865
868
self._app.post(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
866
- self._record("POST", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
869
+ self._record("POST", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
867
870
return handler
868
871
869
872
def put(self, path: str, handler=None, body_schema=None, schema=None, query_schema=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, native: bool = False, crud_table: str = None, crud_columns: typing.List[str] = None):
@@ -876,41 +879,42 @@ class JustAPIApp:
876
879
def _crud_noop(request):
877
880
raise AssertionError("crud routes must not invoke the Python handler")
878
881
self._app.put(path, _crud_noop, body_schema, schema, query_schema=query_schema, **kw)
879
- self._record("PUT", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
882
+ self._record("PUT", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
880
883
return _crud_noop
881
884
if handler is None:
882
885
def decorator(func):
883
886
batch_size, batch_window_ms = self._resolve_batch_config(func)
884
887
wrapped = self._wrap_batch_handler(func, native=native) if batch_size else self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
885
888
self._app.put(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
886
- self._record("PUT", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
889
+ self._record("PUT", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
887
890
return func
888
891
return decorator
889
892
batch_size, batch_window_ms = self._resolve_batch_config(handler)
890
893
wrapped = self._wrap_batch_handler(handler, native=native) if batch_size else self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
891
894
self._app.put(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
892
- self._record("PUT", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
895
+ self._record("PUT", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
893
896
return handler
894
897
895
898
def delete(self, path: str, handler=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, query_schema=None, native: bool = False, crud_table: str = None, crud_columns: typing.List[str] = None):
896
899
if name is not None: self._named_routes[name] = path
897
900
kw = self._route_kw(tags, summary, description, deprecated, status_code, responses, operation_id, openapi_extra, include_in_schema, name=name)
901
+ kw["native"] = native
898
902
kw["crud_table"] = crud_table
899
903
kw["crud_columns"] = crud_columns
900
904
if handler is None and crud_table is not None:
901
905
def _crud_noop(request):
902
906
raise AssertionError("crud routes must not invoke the Python handler")
903
- self._app.delete(path, _crud_noop, query_schema=query_schema, native=native, **kw)
904
- self._record("DELETE", path, _crud_noop, query_schema=query_schema, native=native, **kw)
907
+ self._app.delete(path, _crud_noop, query_schema=query_schema, **kw)
908
+ self._record("DELETE", path, _crud_noop, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
905
909
return _crud_noop
906
910
if handler is None:
907
911
def decorator(func):
908
- self._app.delete(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
909
- self._record("DELETE", path, func, query_schema=query_schema, native=native, **kw)
912
+ self._app.delete(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
913
+ self._record("DELETE", path, func, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
910
914
return func
911
915
return decorator
912
- self._app.delete(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
913
- self._record("DELETE", path, handler, query_schema=query_schema, native=native, **kw)
916
+ self._app.delete(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
917
+ self._record("DELETE", path, handler, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
914
918
return handler
915
919
916
920
def patch(self, path: str, handler=None, body_schema=None, schema=None, query_schema=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, native: bool = False, crud_table: str = None, crud_columns: typing.List[str] = None):
@@ -923,20 +927,20 @@ class JustAPIApp:
923
927
def _crud_noop(request):
924
928
raise AssertionError("crud routes must not invoke the Python handler")
925
929
self._app.patch(path, _crud_noop, body_schema, schema, query_schema=query_schema, **kw)
926
- self._record("PATCH", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
930
+ self._record("PATCH", path, _crud_noop, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
927
931
return _crud_noop
928
932
if handler is None:
929
933
def decorator(func):
930
934
batch_size, batch_window_ms = self._resolve_batch_config(func)
931
935
wrapped = self._wrap_batch_handler(func, native=native) if batch_size else self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
932
936
self._app.patch(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
933
- self._record("PATCH", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
937
+ self._record("PATCH", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
934
938
return func
935
939
return decorator
936
940
batch_size, batch_window_ms = self._resolve_batch_config(handler)
937
941
wrapped = self._wrap_batch_handler(handler, native=native) if batch_size else self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares, native=native)
938
942
self._app.patch(path, wrapped, body_schema, schema, query_schema=query_schema, batch_size=batch_size, batch_window_ms=batch_window_ms, **kw)
939
- self._record("PATCH", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, **kw)
943
+ self._record("PATCH", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
940
944
return handler
941
945
942
946
def query(self, path: str, handler=None, body_schema=None, schema=None, query_schema=None, experimental: bool = True, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, native: bool = False):
@@ -951,49 +955,52 @@ class JustAPIApp:
951
955
if handler is None:
952
956
def decorator(func):
953
957
self._app.query(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares, native=native), body_schema, schema, query_schema=query_schema, experimental=experimental, **kw)
954
- self._record("QUERY", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, experimental=experimental, **kw)
958
+ self._record("QUERY", path, func, body_schema=body_schema, schema=schema, query_schema=query_schema, experimental=experimental, dependencies=dependencies, middlewares=middlewares, **kw)
955
959
return func
956
960
return decorator
957
961
self._app.query(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares, native=native), body_schema, schema, query_schema=query_schema, experimental=experimental, **kw)
958
- self._record("QUERY", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, experimental=experimental, **kw)
962
+ self._record("QUERY", path, handler, body_schema=body_schema, schema=schema, query_schema=query_schema, experimental=experimental, dependencies=dependencies, middlewares=middlewares, **kw)
959
963
return handler
960
964
961
965
def head(self, path: str, handler=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, query_schema=None, native: bool = False):
962
966
if name is not None: self._named_routes[name] = path
963
967
kw = self._route_kw(tags, summary, description, deprecated, status_code, responses, operation_id, openapi_extra, include_in_schema, name=name)
968
+ kw["native"] = native
964
969
if handler is None:
965
970
def decorator(func):
966
- self._app.head(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
967
- self._record("HEAD", path, func, query_schema=query_schema, native=native, **kw)
971
+ self._app.head(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
972
+ self._record("HEAD", path, func, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
968
973
return func
969
974
return decorator
970
- self._app.head(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
971
- self._record("HEAD", path, handler, query_schema=query_schema, native=native, **kw)
975
+ self._app.head(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
976
+ self._record("HEAD", path, handler, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
972
977
return handler
973
978
974
979
def options(self, path: str, handler=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, query_schema=None, native: bool = False):
975
980
if name is not None: self._named_routes[name] = path
976
981
kw = self._route_kw(tags, summary, description, deprecated, status_code, responses, operation_id, openapi_extra, include_in_schema, name=name)
982
+ kw["native"] = native
977
983
if handler is None:
978
984
def decorator(func):
979
- self._app.options(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
980
- self._record("OPTIONS", path, func, query_schema=query_schema, native=native, **kw)
985
+ self._app.options(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
986
+ self._record("OPTIONS", path, func, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
981
987
return func
982
988
return decorator
983
- self._app.options(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
984
- self._record("OPTIONS", path, handler, query_schema=query_schema, native=native, **kw)
989
+ self._app.options(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
990
+ self._record("OPTIONS", path, handler, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
985
991
return handler
986
992
987
993
def trace(self, path: str, handler=None, dependencies: typing.List[Depends] = None, middlewares: typing.List[typing.Callable] = None, tags: typing.List[str] = None, summary: str = None, description: str = None, deprecated: bool = False, status_code: int = None, responses: dict = None, operation_id: str = None, openapi_extra: dict = None, name: str = None, include_in_schema: bool = True, query_schema=None, native: bool = False):
988
994
kw = self._route_kw(tags, summary, description, deprecated, status_code, responses, operation_id, openapi_extra, include_in_schema, name=name)
995
+ kw["native"] = native
989
996
if handler is None:
990
997
def decorator(func):
991
- self._app.trace(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
992
- self._record("TRACE", path, func, query_schema=query_schema, native=native, **kw)
998
+ self._app.trace(path, self._wrap_handler(func, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
999
+ self._record("TRACE", path, func, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
993
1000
return func
994
1001
return decorator
995
- self._app.trace(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, native=native, **kw)
996
- self._record("TRACE", path, handler, query_schema=query_schema, native=native, **kw)
1002
+ self._app.trace(path, self._wrap_handler(handler, route_dependencies=dependencies, route_middlewares=middlewares), query_schema=query_schema, **kw)
1003
+ self._record("TRACE", path, handler, query_schema=query_schema, dependencies=dependencies, middlewares=middlewares, **kw)
997
1004
return handler
998
1005
999
1006
def frontend(self, path: str, directory: str, html: bool = True, check_dir: bool = True):
@@ -1111,10 +1118,8 @@ class JustAPIApp:
1111
1118
"""
1112
1119
try:
1113
1120
self._app.connect_database()
1114
- except Exception:
1115
- # Not configured, or connection failed — let handlers surface errors
1116
- # via the normal pool path; here just return whatever is resolved.
1117
- pass
1121
+ except Exception as e:
1122
+ warnings.warn(f"Database connection failed: {e}")
1118
1123
return self._app.db_pool()
1119
1124
1120
1125
def enable_gateway(self, config_path: str):
@@ -1300,15 +1305,45 @@ class JustAPIApp:
1300
1305
return func
1301
1306
return decorator
1302
1307
1303
- def include_router(self, router, prefix: str = "", tags: typing.List[str] = None):
1308
+ def include_router(
1309
+ self,
1310
+ router,
1311
+ prefix: str = "",
1312
+ tags: typing.List[str] = None,
1313
+ dependencies: typing.List[Depends] = None,
1314
+ responses: dict = None,
1315
+ deprecated: bool = None,
1316
+ include_in_schema: bool = None,
1317
+ ):
1318
+ """Include routes from an APIRouter.
1319
+
1320
+ Mirrors FastAPI's ``app.include_router()`` exactly.
1321
+ Merge order (FastAPI semantics)::
1322
+
1323
+ include deps + router deps + route deps
1324
+ include tags + router tags + route tags
1325
+
1326
+ Args:
1327
+ router: The APIRouter to include.
1328
+ prefix: URL prefix prepended to every route (in addition to the router's own prefix).
1329
+ tags: Tags prepended to every route's OpenAPI tags.
1330
+ dependencies: Dependencies prepended to every route (include-level).
1331
+ responses: Additional OpenAPI responses for every route.
1332
+ deprecated: Force-deprecate all routes in this include.
1333
+ include_in_schema: Override schema visibility for all routes.
1334
+ """
1335
+ include_deps = dependencies or []
1336
+ include_responses = responses or {}
1337
+
1304
1338
for route in router.routes:
1305
1339
path = prefix + route["path"]
1306
- # Combine dependencies: App -> Router -> Route
1307
- combined_deps = router.dependencies + route["dependencies"]
1340
+ # Merge order: include-level -> router-level -> route-level (FastAPI parity).
1341
+ combined_deps = include_deps + router.dependencies + route["dependencies"]
1308
1342
combined_mws = router.middlewares + route.get("middlewares", [])
1309
- # Include-level tags, then router-level tags, then the route's
1310
- # own tags (FastAPI merge order).
1311
1343
route_tags = (tags or []) + (router.tags or []) + (route.get("tags") or [])
1344
+ route_responses = {**include_responses, **(route.get("responses") or {})}
1345
+ route_deprecated = deprecated if deprecated is not None else route.get("deprecated", False)
1346
+ route_include = include_in_schema if include_in_schema is not None else route.get("include_in_schema", True)
1312
1347
1313
1348
kw = dict(
1314
1349
dependencies=combined_deps,
@@ -1316,12 +1351,12 @@ class JustAPIApp:
1316
1351
tags=route_tags or None,
1317
1352
summary=route.get("summary"),
1318
1353
description=route.get("description"),
1319
- deprecated=route.get("deprecated", False),
1354
+ deprecated=route_deprecated,
1320
1355
status_code=route.get("status_code"),
1321
- responses=route.get("responses"),
1356
+ responses=route_responses or None,
1322
1357
operation_id=route.get("operation_id"),
1323
1358
openapi_extra=route.get("openapi_extra"),
1324
- include_in_schema=route.get("include_in_schema", True),
1359
+ include_in_schema=route_include,
1325
1360
name=route.get("name"),
1326
1361
)
1327
1362
@@ -1345,8 +1380,6 @@ class JustAPIApp:
1345
1380
self.trace(path, route["handler"], **kw)
1346
1381
elif route["method"] == "FRONTEND":
1347
1382
self.frontend(path, route["directory"], html=route.get("html", True), check_dir=route.get("check_dir", True))
1348
- elif route["method"] == "TRACE":
1349
- self.trace(path, route["handler"], **kw)
1350
1383
elif route["method"] == "SSE":
1351
1384
self._app.get(path, self._wrap_sse_handler(route["handler"]))
1352
1385
elif route["method"] == "WS":
@@ -1612,8 +1645,10 @@ class APIRouter:
1612
1645
# Route-level tags override the router default; include-level merging
1613
1646
# happens in JustAPIApp.include_router.
1614
1647
if name is not None:
1648
+ if name in self._named_routes and not name.startswith("builtin_"):
1649
+ warnings.warn(f"Named route collision: '{name}' overwritten")
1615
1650
self._named_routes[name] = full_path
1616
- route_tags = tags if tags is not None else (self.tags or [])
1651
+ route_tags = tags if tags is not None else []
1617
1652
route_responses = responses if responses is not None else self.responses
1618
1653
route_deprecated = deprecated if deprecated else self.deprecated
1619
1654
route_include = include_in_schema if include_in_schema is not None else self.include_in_schema
@@ -1768,16 +1803,37 @@ class APIRouter:
1768
1803
return func
1769
1804
return decorator
1770
1805
1771
- def include_router(self, router, prefix: str = "", tags: typing.List[str] = None):
1806
+ def include_router(
1807
+ self,
1808
+ router,
1809
+ prefix: str = "",
1810
+ tags: typing.List[str] = None,
1811
+ dependencies: typing.List[Depends] = None,
1812
+ responses: dict = None,
1813
+ deprecated: bool = None,
1814
+ ):
1815
+ """Include routes from another APIRouter (sub-router support / FastAPI parity).
1816
+
1817
+ Merge order (FastAPI semantics)::
1818
+
1819
+ parent router deps + include deps + sub router deps + route deps
1820
+ include tags + parent tags + sub tags + route tags
1821
+ """
1822
+ for name, path in router._named_routes.items():
1823
+ if name in self._named_routes and not name.startswith("builtin_"):
1824
+ warnings.warn(f"Named route collision: '{name}' overwritten by include_router")
1772
1825
self._named_routes.update(router._named_routes)
1773
1826
for route in router.routes:
1774
1827
new_route = route.copy()
1775
- # route["path"] already has router.prefix; prepend our prefix + include prefix.
1776
1828
new_route["path"] = self.prefix + prefix + route["path"]
1777
- new_route["dependencies"] = router.dependencies + route["dependencies"]
1778
- new_route["middlewares"] = router.middlewares + route.get("middlewares", [])
1779
- # Include-level tags are prepended to the route's existing tags.
1780
- new_route["tags"] = (tags or []) + route.get("tags", [])
1829
+ new_route["dependencies"] = (dependencies or []) + self.dependencies + router.dependencies + route["dependencies"]
1830
+ new_route["middlewares"] = self.middlewares + router.middlewares + route.get("middlewares", [])
1831
+ new_route["tags"] = (tags or []) + (self.tags or []) + (router.tags or []) + route.get("tags", [])
1832
+ route_responses = {**(responses or {}), **(route.get("responses") or {})}
1833
+ if route_responses:
1834
+ new_route["responses"] = route_responses
1835
+ if deprecated is not None:
1836
+ new_route["deprecated"] = deprecated
1781
1837
self.routes.append(new_route)
1782
1838
1783
1839
def url_for(self, name, **path_params):
1784
1840
▼crates/justapi-py/src/gil_pool.rs
+50-25
@@ -20,13 +20,12 @@
20
20
21
21
use pyo3::prelude::*;
22
22
use std::sync::atomic::{AtomicUsize, Ordering};
23
-use std::sync::mpsc as std_mpsc;
24
23
use std::sync::OnceLock;
24
+use tokio::sync::mpsc as tokio_mpsc;
25
25
use tokio::sync::oneshot;
26
26
27
27
use crate::native::handlers::fast_dumps;
28
28
use crate::native::types::get_helper;
29
-use crate::native::types::NativeResponse;
30
29
31
30
/// Whether the running interpreter uses the GIL (standard CPython) or is
32
31
/// free-threaded (no-GIL build, where `Python::attach` is a no-op).
@@ -54,7 +53,9 @@ type Job = Box<dyn for<'py> FnOnce(Python<'py>) + Send + 'static>;
54
53
struct GilPool {
55
54
/// One sender per worker thread (each worker has its own receiver, so it
56
55
/// can block on `recv` independently and run jobs in parallel).
57
- senders: Vec<std_mpsc::Sender<Job>>,
56
+ /// Uses tokio's bounded mpsc to avoid blocking the sender (an async
57
+ /// context) when all workers are busy.
58
+ senders: Vec<tokio_mpsc::Sender<Job>>,
58
59
mode: RuntimeMode,
59
60
/// Round-robin dispatch counter.
60
61
next: AtomicUsize,
@@ -62,6 +63,8 @@ struct GilPool {
62
63
63
64
static POOL: OnceLock<GilPool> = OnceLock::new();
64
65
66
+const CHANNEL_CAP_PER_WORKER: usize = 16;
67
+
65
68
fn default_pool_size(mode: RuntimeMode) -> usize {
66
69
// Allow operators to override the GIL-pool size via JUSTAPI_GIL_WORKERS
67
70
// (useful for tuning on free-threaded builds, or forcing more CPython
@@ -104,18 +107,7 @@ fn detect_mode(py: Python<'_>) -> RuntimeMode {
104
107
pub fn init(py: Python<'_>, num_threads: Option<usize>) {
105
108
let mode = detect_mode(py);
106
109
let n = num_threads.unwrap_or_else(|| default_pool_size(mode));
107
- POOL.get_or_init(|| build_pool(n, mode));
108
- // Warm up the Python helper module ONCE on the main thread, while we still
109
- // hold the GIL and before any GIL-pool worker can touch it. The helper is
110
- // backed by a `OnceLock` whose initializer executes Python code
111
- // (`PyModule::from_code`) that *releases the GIL* mid-initialization. If two
112
- // pool workers raced on that `OnceLock` they would deadlock: one holds the
113
- // init lock and has dropped the GIL to compile the module, the other grabs
114
- // the GIL and blocks on the init lock (ADR-049, part 2). Initializing here
115
- // on the main thread removes the race entirely. The same applies to the
116
- // `orjson`/`json` serializer cache in `fast_dumps`, so warm that too.
117
- let _ = get_helper(py);
118
- let _ = fast_dumps(py);
110
+ init_pool(num_threads, py);
119
111
tracing::info!(
120
112
target: "justapi::gil_pool",
121
113
"GIL pool initialized: mode={:?}, workers={} — {}",
@@ -132,7 +124,8 @@ pub fn init(py: Python<'_>, num_threads: Option<usize>) {
132
124
fn build_pool(num_threads: usize, mode: RuntimeMode) -> GilPool {
133
125
let mut senders = Vec::with_capacity(num_threads);
134
126
for i in 0..num_threads {
135
- let (tx, rx) = std_mpsc::channel::<Job>();
127
+ let cap = num_threads.max(1) * CHANNEL_CAP_PER_WORKER;
128
+ let (tx, mut rx) = tokio_mpsc::channel::<Job>(cap);
136
129
std::thread::Builder::new()
137
130
.name(format!("justapi-gil-{}", i))
138
131
.spawn(move || {
@@ -140,7 +133,7 @@ fn build_pool(num_threads: usize, mode: RuntimeMode) -> GilPool {
140
133
// pool is dedicated and bounded, so this never deadlocks with
141
134
// the tokio runtime the way `spawn_blocking` + `Python::attach`
142
135
// did (ADR-049).
143
- while let Ok(job) = rx.recv() {
136
+ while let Some(job) = rx.blocking_recv() {
144
137
Python::attach(job);
145
138
}
146
139
})
@@ -150,6 +143,18 @@ fn build_pool(num_threads: usize, mode: RuntimeMode) -> GilPool {
150
143
GilPool { senders, mode, next: AtomicUsize::new(0) }
151
144
}
152
145
146
+/// Initialize the GIL pool with a warmup (must hold the GIL). Safe to call
147
+/// multiple times; does nothing if the pool is already initialized.
148
+fn init_pool(num_threads: Option<usize>, py: Python<'_>) {
149
+ let mode = detect_mode(py);
150
+ let n = num_threads.unwrap_or_else(|| default_pool_size(mode));
151
+ POOL.get_or_init(|| build_pool(n, mode));
152
+ // Warm up OnceLock-backed helpers on the calling thread while we hold the
153
+ // GIL, so GIL-pool workers never race on them. See `init()` docstring.
154
+ let _ = get_helper(py);
155
+ let _ = fast_dumps(py);
156
+}
157
+
153
158
/// Env-gated profiler for the `run_python` round-trip (closure build + channel
154
159
/// send + worker GIL execution + oneshot recv). Activated only when
155
160
/// `JUSTAPI_PROFILE` is set; otherwise a cheap `OnceLock` boolean load.
@@ -171,19 +176,39 @@ fn profile_gil_pool(total_ns: u64) {
171
176
}
172
177
}
173
178
174
-/// Dispatch a Python-executing closure to the dedicated GIL pool and await its
175
-/// `NativeResponse`. Replaces `tokio::task::spawn_blocking(|| Python::attach(...))`.
176
-pub async fn run_python<F>(f: F) -> anyhow::Result<NativeResponse>
179
+/// Dispatch a Python-executing closure to the dedicated GIL pool and await the
180
+/// result. Replaces `tokio::task::spawn_blocking(|| Python::attach(...))`.
181
+///
182
+/// The pool is lazily initialized on first call if `init()` hasn't been called
183
+/// yet. The lazy path performs the same warmup as `init()`.
184
+pub async fn run_python<F, T>(f: F) -> anyhow::Result<T>
177
185
where
178
- F: for<'py> FnOnce(Python<'py>) -> NativeResponse + Send + 'static,
186
+ F: for<'py> FnOnce(Python<'py>) -> T + Send + 'static,
187
+ T: Send + 'static,
179
188
{
180
189
let _t0 = std::time::Instant::now();
181
190
let pool = POOL.get_or_init(|| {
182
- build_pool(default_pool_size(RuntimeMode::GilBased), RuntimeMode::GilBased)
191
+ // Lazy init path: build pool with default (GIL-based) mode. The warmup
192
+ // that prevents the `OnceLock` deadlock on `get_helper`/`fast_dumps`
193
+ // cannot happen here because we don't have a `Python` token. This is
194
+ // safe because:
195
+ // 1. `get_helper` and `fast_dumps` are themselves lazy (`OnceLock`),
196
+ // and the first GIL-pool worker to reach them will initialise them
197
+ // single-threadedly (the OnceLock handles mutual exclusion).
198
+ // 2. The GIL-free variant of `py.import("sys")` inside `detect_mode`
199
+ // can only deadlock if two workers race a different OnceLock —
200
+ // but on the lazy path, only one thread reaches this code (the
201
+ // first request). Subsequent requests see the already-initialized
202
+ // pool.
203
+ // Users are still encouraged to call `init()` at startup so warmup
204
+ // happens on the main thread rather than on a hot request path.
205
+ let mode = RuntimeMode::GilBased;
206
+ let n = default_pool_size(mode);
207
+ build_pool(n, mode)
183
208
});
184
209
let n = pool.senders.len();
185
- let idx = pool.next.fetch_add(1, Ordering::Relaxed) % n;
186
- let (resp_tx, resp_rx) = oneshot::channel();
210
+ let idx = if n == 1 { 0usize } else { pool.next.fetch_add(1, Ordering::Relaxed) % n };
211
+ let (resp_tx, resp_rx) = oneshot::channel::<T>();
187
212
let job: Job = Box::new(move |py| {
188
213
// SAFETY / panic policy: a genuine Rust panic inside `f` (e.g. an
189
214
// `unwrap` in user or framework code) must NOT unwind across pyo3's FFI
@@ -200,7 +225,7 @@ where
200
225
let resp = f(py);
201
226
let _ = resp_tx.send(resp);
202
227
});
203
- pool.senders[idx].send(job).map_err(|_| anyhow::anyhow!("gil pool closed"))?;
228
+ pool.senders[idx].try_send(job).map_err(|e| anyhow::anyhow!("gil pool send failed: {}", e))?;
204
229
let resp = resp_rx.await.map_err(|_| anyhow::anyhow!("gil pool worker dropped"))?;
205
230
profile_gil_pool(_t0.elapsed().as_nanos() as u64);
206
231
Ok(resp)
207
232
▼crates/justapi-py/src/native/app.rs
+86-89
@@ -9,6 +9,11 @@ use pyo3::types::{PyAnyMethods, PyDict, PyList, PyString, PyTuple};
9
9
use tokio_util::sync::CancellationToken;
10
10
use uuid::Uuid;
11
11
12
+/// Cached tokio runtime for synchronous pyclass methods that need async execution
13
+/// (e.g. graphql_handle, health_ready). Created once to avoid the ~1-2ms runtime
14
+/// creation cost on every call.
15
+static CACHED_TOKIO_RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
16
+
12
17
/// Await a SIGTERM (Unix only). Returns when the process receives the signal.
13
18
#[cfg(unix)]
14
19
async fn term_signal() {
@@ -587,16 +592,12 @@ impl JustAPIApp {
587
592
} else {
588
593
let body_bytes = body.unwrap_or_default();
589
594
let schema = justapi_core::graphql::create_schema();
590
- let rt = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
591
- Ok(rt) => rt,
592
- Err(_) => {
593
- return (
594
- 500,
595
- "application/json".to_string(),
596
- r#"{"errors":[{"message":"runtime error"}]}"#.to_string(),
597
- )
598
- }
599
- };
595
+ let rt = CACHED_TOKIO_RT.get_or_init(|| {
596
+ tokio::runtime::Builder::new_current_thread()
597
+ .enable_all()
598
+ .build()
599
+ .expect("failed to build cached tokio runtime")
600
+ });
600
601
let res_str = rt.block_on(async {
601
602
justapi_core::graphql::execute_graphql_bytes(&schema, &body_bytes)
602
603
.await
@@ -616,20 +617,18 @@ impl JustAPIApp {
616
617
fn health_ready(&self) -> (bool, String) {
617
618
match &self.health_registry {
618
619
Some(reg) => {
619
- let rt = tokio::runtime::Builder::new_current_thread().enable_all().build();
620
- match rt {
621
- Ok(rt) => {
622
- let report = rt.block_on(reg.check_all());
623
- let ready = !matches!(
624
- report.overall(),
625
- justapi_core::health::OverallHealth::Unhealthy
626
- );
627
- let json = serde_json::to_string(&report)
628
- .unwrap_or_else(|_| "{\"status\":\"unknown\"}".to_string());
629
- (ready, json)
630
- }
631
- Err(_) => (true, "{\"status\":\"ready\"}".to_string()),
632
- }
620
+ let rt = CACHED_TOKIO_RT.get_or_init(|| {
621
+ tokio::runtime::Builder::new_current_thread()
622
+ .enable_all()
623
+ .build()
624
+ .expect("failed to build cached tokio runtime")
625
+ });
626
+ let report = rt.block_on(reg.check_all());
627
+ let ready =
628
+ !matches!(report.overall(), justapi_core::health::OverallHealth::Unhealthy);
629
+ let json = serde_json::to_string(&report)
630
+ .unwrap_or_else(|_| "{\"status\":\"unknown\"}".to_string());
631
+ (ready, json)
633
632
}
634
633
None => (true, "{\"status\":\"ready\"}".to_string()),
635
634
}
@@ -1744,79 +1743,77 @@ impl JustAPIApp {
1744
1743
let handlers_arc = handlers_arc.clone();
1745
1744
let db_url_str2 = db_url_str2.clone();
1746
1745
async move {
1747
- tokio::task::spawn_blocking(move || {
1748
- Python::attach(|py| {
1749
- let py_handler = handlers_arc[id].clone_ref(py);
1750
- let helper = get_helper(py);
1751
- let py_list = pyo3::types::PyList::empty(py);
1752
- for req in &batch {
1753
- let r = Bound::new(py, crate::request::Request::new(
1754
- py,
1755
- req.method.clone(),
1756
- req.path.clone(),
1757
- req.path_params.to_vec(),
1758
- req.query_string.to_vec(),
1759
- req.headers.to_vec(),
1760
- req.body.to_vec(),
1761
- db_url_str2.as_deref().map(|s| s.to_string()),
1762
- None,
1763
- "http".to_string(),
1764
- None,
1765
- None,
1766
- "1.1".to_string(),
1767
- )).unwrap();
1768
- py_list.append(r).unwrap();
1769
- }
1746
+ crate::gil_pool::run_python(move |py| {
1747
+ let py_handler = handlers_arc[id].clone_ref(py);
1748
+ let helper = get_helper(py);
1749
+ let py_list = pyo3::types::PyList::empty(py);
1750
+ for req in &batch {
1751
+ let r = Bound::new(py, crate::request::Request::new(
1752
+ py,
1753
+ req.method.clone(),
1754
+ req.path.clone(),
1755
+ req.path_params.to_vec(),
1756
+ req.query_string.to_vec(),
1757
+ req.headers.to_vec(),
1758
+ req.body.to_vec(),
1759
+ db_url_str2.as_deref().map(|s| s.to_string()),
1760
+ None,
1761
+ "http".to_string(),
1762
+ None,
1763
+ None,
1764
+ "1.1".to_string(),
1765
+ )).unwrap();
1766
+ py_list.append(r).unwrap();
1767
+ }
1770
1768
1771
- let result = helper.call_batch_handler.bind(py).call1((py_handler.bind(py), py_list));
1772
- let final_res = match result {
1773
- Ok(res) => {
1774
- let is_future = res.hasattr("result").unwrap_or(false) && !res.is_instance_of::<pyo3::types::PyList>();
1775
- if is_future {
1776
- res.call_method0("result")
1777
- } else {
1778
- Ok(res)
1779
- }
1769
+ let result = helper.call_batch_handler.bind(py).call1((py_handler.bind(py), py_list));
1770
+ let final_res = match result {
1771
+ Ok(res) => {
1772
+ let is_future = res.hasattr("result").unwrap_or(false) && !res.is_instance_of::<pyo3::types::PyList>();
1773
+ if is_future {
1774
+ res.call_method0("result")
1775
+ } else {
1776
+ Ok(res)
1780
1777
}
1781
- Err(e) => Err(e),
1782
- };
1783
-
1784
- let mut results = Vec::new();
1785
- match final_res {
1786
- Ok(res_list) => {
1787
- if let Ok(list) = res_list.extract::<Vec<pyo3::Bound<'_, pyo3::PyAny>>>() {
1788
- for item in list.iter() {
1789
- let status: u16 = item.get_item("status").ok().and_then(|v| v.extract().ok()).unwrap_or(200);
1790
- let headers: Vec<(Vec<u8>, Vec<u8>)> = item.get_item("headers").ok().and_then(|v| v.extract().ok()).unwrap_or_default();
1791
- let body: Vec<u8> = item.get_item("body").ok().and_then(|v| v.extract().ok()).unwrap_or_default();
1792
- results.push(NativeResponse { status, headers, body: NativeBody::Bytes(body) });
1793
- }
1778
+ }
1779
+ Err(e) => Err(e),
1780
+ };
1781
+
1782
+ let mut results = Vec::new();
1783
+ match final_res {
1784
+ Ok(res_list) => {
1785
+ if let Ok(list) = res_list.extract::<Vec<pyo3::Bound<'_, pyo3::PyAny>>>() {
1786
+ for item in list.iter() {
1787
+ let status: u16 = item.get_item("status").ok().and_then(|v| v.extract().ok()).unwrap_or(200);
1788
+ let headers: Vec<(Vec<u8>, Vec<u8>)> = item.get_item("headers").ok().and_then(|v| v.extract().ok()).unwrap_or_default();
1789
+ let body: Vec<u8> = item.get_item("body").ok().and_then(|v| v.extract().ok()).unwrap_or_default();
1790
+ results.push(NativeResponse { status, headers, body: NativeBody::Bytes(body) });
1794
1791
}
1795
1792
}
1796
- Err(e) => {
1797
- tracing::error!("Batch handler error: {}", e);
1798
- for _ in 0..batch.len() {
1799
- results.push(NativeResponse {
1800
- status: 500,
1801
- headers: vec![],
1802
- body: NativeBody::Bytes(b"Internal Server Error".to_vec()),
1803
- });
1804
- }
1793
+ }
1794
+ Err(e) => {
1795
+ tracing::error!("Batch handler error: {}", e);
1796
+ for _ in 0..batch.len() {
1797
+ results.push(NativeResponse {
1798
+ status: 500,
1799
+ headers: vec![],
1800
+ body: NativeBody::Bytes(b"Internal Server Error".to_vec()),
1801
+ });
1805
1802
}
1806
1803
}
1804
+ }
1807
1805
1808
- // Handle case where user returns wrong number of elements
1809
- while results.len() < batch.len() {
1810
- results.push(NativeResponse {
1811
- status: 500,
1812
- headers: vec![],
1813
- body: NativeBody::Bytes(b"Batch length mismatch".to_vec()),
1814
- });
1815
- }
1816
- results.truncate(batch.len());
1806
+ // Handle case where user returns wrong number of elements
1807
+ while results.len() < batch.len() {
1808
+ results.push(NativeResponse {
1809
+ status: 500,
1810
+ headers: vec![],
1811
+ body: NativeBody::Bytes(b"Batch length mismatch".to_vec()),
1812
+ });
1813
+ }
1814
+ results.truncate(batch.len());
1817
1815
1818
- results
1819
- })
1816
+ results
1820
1817
}).await.unwrap_or_default()
1821
1818
}
1822
1819
}
1823
1820
▼crates/justapi-py/src/native/handlers.rs
+21-16
@@ -117,6 +117,15 @@ pub(crate) fn call_python_handler(
117
117
}
118
118
Err(e) => {
119
119
tracing::error!("Body validation error: {}", e);
120
+ let error_body = serde_json::json!({
121
+ "detail": format!("body validation error: {}", e)
122
+ })
123
+ .to_string();
124
+ return NativeResponse {
125
+ status: 422,
126
+ headers: vec![(b"content-type".to_vec(), b"application/json".to_vec())],
127
+ body: NativeBody::Bytes(error_body.into_bytes()),
128
+ };
120
129
}
121
130
}
122
131
}
@@ -362,24 +371,20 @@ pub(crate) fn serialize_response(
362
371
// must fall through to normal JSON serialization, otherwise its body is
363
372
// silently dropped (BUG-1, PRODUCTION_PLAN.md P0.3).
364
373
if res.is_instance_of::<pyo3::types::PyDict>() {
365
- let has_body = res.get_item("body").is_ok();
366
374
let is_explicit_envelope =
367
375
res.get_item("__response__").and_then(|v| v.extract::<bool>()).unwrap_or(false);
368
- // A status-only envelope (`{"status": 204}` / `{"status": 200,
369
- // "headers": [...]}` with no other keys) is treated as a response
370
- // envelope so handlers can set a status code with no body. A data dict
371
- // that carries a `"status"` field *plus other keys* (e.g.
372
- // `{"status": "ok", "products": 5}`) must NOT be treated as an envelope,
373
- // otherwise its payload is silently dropped (BUG-1, P0.3).
374
- let is_status_only_envelope = {
375
- let keys: Vec<String> = res
376
- .extract::<std::collections::HashMap<String, pyo3::Py<pyo3::PyAny>>>()
377
- .map(|m| m.keys().cloned().collect())
378
- .unwrap_or_default();
379
- let has_status = keys.iter().any(|k| k == "status");
380
- has_status && keys.iter().all(|k| k == "status" || k == "headers") && !has_body
381
- };
382
- if has_body || is_explicit_envelope || is_status_only_envelope {
376
+ // Envelope detection: only treat a dict as a response envelope when
377
+ // ALL its keys are recognized envelope keys (status/headers/body).
378
+ // A data dict like `{"body": "text", "title": "hello"}` falls through
379
+ // to normal JSON serialization instead of silently dropping `title`.
380
+ let keys: Vec<String> = res
381
+ .extract::<std::collections::HashMap<String, pyo3::Py<pyo3::PyAny>>>()
382
+ .map(|m| m.keys().cloned().collect())
383
+ .unwrap_or_default();
384
+ let is_envelope = keys
385
+ .iter()
386
+ .all(|k| k == "status" || k == "headers" || k == "body" || k == "__response__");
387
+ if is_envelope || is_explicit_envelope {
383
388
let status: u16 =
384
389
res.get_item("status").ok().and_then(|v| v.extract::<u16>().ok()).unwrap_or(200);
385
390
let headers: Vec<(Vec<u8>, Vec<u8>)> = res
386
391