swadhin / justapi public

swadhinbiswas

[email protected]

2026-07-08 08:30:00 +0600 committed NaN years ago

243e26c

commit 243e26c

16 files changed

+2540 -0

Files

crates

justapi-py

src

native

app.rs +663 handlers.rs +651 mod.rs +9 types.rs +138 buffer_test.rs +59 dag.rs +233 database.rs +59 lib.rs +61 multipart.rs +38 rate_limit.rs +59 request.rs +166 test_async.rs +1 test_client.rs +177 websocket.rs +141 Cargo.toml +26 pyproject.toml +59

Showing 16 changed files+2540-0

▼crates/justapi-py/Cargo.toml

+26-0

@@ -0,0 +1,26 @@
	1	+[package]
	2	+name = "justapi-py"
	3	+version.workspace = true
	4	+edition.workspace = true
	5	+license.workspace = true
	6	+
	7	+[lib]
	8	+crate-type = ["cdylib", "lib"]
	9	+
	10	+[dependencies]
	11	+pyo3.workspace = true
	12	+justapi-core = { workspace = true, features = ["db", "ws"] }
	13	+tokio.workspace = true
	14	+http-body-util.workspace = true
	15	+hyper.workspace = true
	16	+http = "1"
	17	+futures = "0.3"
	18	+serde.workspace = true
	19	+serde_json.workspace = true
	20	+tracing.workspace = true
	21	+anyhow.workspace = true
	22	+tokio-util.workspace = true
	23	+tonic = "0.14.6"
	24	+tokio-stream.workspace = true
	25	+tokio-tungstenite = "0.24"
	26	+pyo3-async-runtimes = { version = "0.29.0", features = ["tokio", "tokio-runtime"] }
0	27

▼crates/justapi-py/pyproject.toml

+59-0

@@ -0,0 +1,59 @@
	1	+[build-system]
	2	+requires = ["maturin>=1.0,<2.0"]
	3	+build-backend = "maturin"
	4	+
	5	+[project]
	6	+name = "justapi"
	7	+version = "2.0.0"
	8	+description = "JustAPI Runtime — a Rust-powered, FastAPI-class web framework for Python. Blazing-fast routing, native JSON validation, dependency injection, OpenAPI, WebSockets, SSE, background tasks, and an embedded DAG engine — all from a single pip install."
	9	+readme = "README.md"
	10	+requires-python = ">=3.11"
	11	+license = { file = "LICENSE" }
	12	+authors = [{ name = "JustAPI contributors" }]
	13	+keywords = [
	14	+ "web",
	15	+ "framework",
	16	+ "api",
	17	+ "fastapi",
	18	+ "rust",
	19	+ "pyo3",
	20	+ "async",
	21	+ "openapi",
	22	+ "websocket",
	23	+ "rest",
	24	+]
	25	+classifiers = [
	26	+ "Development Status :: 5 - Production/Stable",
	27	+ "Intended Audience :: Developers",
	28	+ "License :: OSI Approved :: MIT License",
	29	+ "Programming Language :: Python :: 3",
	30	+ "Programming Language :: Python :: 3.11",
	31	+ "Programming Language :: Python :: 3.12",
	32	+ "Programming Language :: Python :: 3.13",
	33	+ "Programming Language :: Python :: 3.14",
	34	+ "Programming Language :: Rust",
	35	+ "Topic :: Internet :: WWW/HTTP",
	36	+ "Topic :: Internet :: WWW/HTTP :: HTTP Servers",
	37	+ "Topic :: Software Development :: Libraries :: Application Frameworks",
	38	+ "Typing :: Typed",
	39	+]
	40	+dependencies = []
	41	+
	42	+[project.optional-dependencies]
	43	+pydantic = ["pydantic>=2.0"]
	44	+jinja = ["jinja2>=3.0"]
	45	+full = ["pydantic>=2.0", "jinja2>=3.0"]
	46	+
	47	+[project.urls]
	48	+Homepage = "https://github.com/anomalyco/justapi"
	49	+Repository = "https://github.com/anomalyco/justapi"
	50	+Documentation = "https://github.com/anomalyco/justapi#readme"
	51	+"Bug Tracker" = "https://github.com/anomalyco/justapi/issues"
	52	+
	53	+[tool.maturin]
	54	+module-name = "justapi._justapi"
	55	+python-source = "python"
	56	+manifest-path = "Cargo.toml"
	57	+strip = true
	58	+abi3-py311 = true
	59	+include = ["README.md", "LICENSE"]
0	60

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

+59-0

@@ -0,0 +1,59 @@
	1	+use pyo3::ffi;
	2	+use pyo3::prelude::*;
	3	+use std::os::raw::c_int;
	4	+
	5	+#[pyclass]
	6	+pub struct ZeroCopyBuffer {
	7	+ data: Vec<u8>,
	8	+}
	9	+
	10	+#[pymethods]
	11	+impl ZeroCopyBuffer {
	12	+ #[new]
	13	+ pub fn new(data: Vec<u8>) -> Self {
	14	+ ZeroCopyBuffer { data }
	15	+ }
	16	+
	17	+ unsafe fn __getbuffer__(
	18	+ slf: PyRefMut<'_, Self>,
	19	+ view: *mut ffi::Py_buffer,
	20	+ _flags: c_int,
	21	+ ) -> PyResult<()> {
	22	+ // SAFETY: The `view` pointer is checked for null before any
	23	+ // dereference (line 22-23). `Py_NewRef` is a safe FFI call that
	24	+ // increments the reference count of the Python object. The `buf`
	25	+ // pointer points into `slf.data` which is a `Vec<u8>` and remains
	26	+ // valid for the lifetime of the `ZeroCopyBuffer`. All other fields
	27	+ // (`readonly`, `itemsize`, `ndim`, `format`, `shape`, `strides`,
	28	+ // `suboffsets`, `internal`) are set to safe defaults that indicate
	29	+ // a flat, read-only, one-dimensional buffer of bytes. The `unsafe`
	30	+ // qualifier is required by PyO3's buffer-protocol trait.
	31	+ if view.is_null() {
	32	+ return Err(pyo3::exceptions::PyBufferError::new_err("View is null"));
	33	+ }
	34	+
	35	+ (*view).obj = ffi::Py_NewRef(slf.as_ptr());
	36	+ (view).buf = slf.data.as_ptr() as mut _;
	37	+ (*view).len = slf.data.len() as isize;
	38	+ (*view).readonly = 1;
	39	+ (*view).itemsize = 1;
	40	+
	41	+ (*view).format = std::ptr::null_mut();
	42	+ (*view).ndim = 0;
	43	+ (*view).shape = std::ptr::null_mut();
	44	+ (*view).strides = std::ptr::null_mut();
	45	+ (*view).suboffsets = std::ptr::null_mut();
	46	+ (*view).internal = std::ptr::null_mut();
	47	+
	48	+ Ok(())
	49	+ }
	50	+
	51	+ unsafe fn __releasebuffer__(&self, _view: *mut ffi::Py_buffer) {
	52	+ // SAFETY: The `__releasebuffer__` is a no-op because no heap
	53	+ // allocations were made inside `__getbuffer__` that require
	54	+ // explicit cleanup (the buffer view directly references the
	55	+ // `Vec<u8>` owned by `ZeroCopyBuffer` which is dropped by Rust's
	56	+ // normal destructor). The `unsafe` qualifier is required by
	57	+ // PyO3's buffer-protocol trait.
	58	+ }
	59	+}
0	60

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

+233-0

@@ -0,0 +1,233 @@
	1	+use pyo3::prelude::*;
	2	+use pyo3::types::{PyDict, PyTuple};
	3	+use std::collections::{HashMap, HashSet};
	4	+use std::sync::Arc;
	5	+use tokio::sync::Notify;
	6	+use tokio_util::sync::CancellationToken;
	7	+
	8	+#[pyclass(name = "DagNode", from_py_object)]
	9	+pub struct DagNode {
	10	+ pub name: String,
	11	+ pub handler: Py<PyAny>,
	12	+ pub dependencies: Vec<String>,
	13	+}
	14	+
	15	+impl Clone for DagNode {
	16	+ fn clone(&self) -> Self {
	17	+ Python::<'_>::try_attach(\|py\| Self {
	18	+ name: self.name.clone(),
	19	+ handler: self.handler.clone_ref(py),
	20	+ dependencies: self.dependencies.clone(),
	21	+ })
	22	+ .expect("GIL required for Clone")
	23	+ }
	24	+}
	25	+
	26	+#[pymethods]
	27	+impl DagNode {
	28	+ #[new]
	29	+ #[pyo3(signature = (name, handler, dependencies=None))]
	30	+ fn new(name: String, handler: Py<PyAny>, dependencies: Option<Vec<String>>) -> Self {
	31	+ Self {
	32	+ name,
	33	+ handler,
	34	+ dependencies: dependencies.unwrap_or_default(),
	35	+ }
	36	+ }
	37	+}
	38	+
	39	+#[pyclass(name = "Dag", from_py_object)]
	40	+#[derive(Clone)]
	41	+pub struct Dag {
	42	+ nodes: HashMap<String, DagNode>,
	43	+}
	44	+
	45	+#[pymethods]
	46	+impl Dag {
	47	+ #[new]
	48	+ fn new(nodes: Vec<DagNode>) -> PyResult<Self> {
	49	+ let mut map = HashMap::new();
	50	+ for node in nodes {
	51	+ map.insert(node.name.clone(), node);
	52	+ }
	53	+ Ok(Self { nodes: map })
	54	+ }
	55	+
	56	+ /// This function returns an `asyncio.Future` in Python.
	57	+ fn execute<'py>(
	58	+ &self,
	59	+ py: Python<'py>,
	60	+ inputs: &Bound<'py, PyDict>,
	61	+ ) -> PyResult<Bound<'py, PyAny>> {
	62	+ let nodes = self.nodes.clone();
	63	+
	64	+ let mut state: HashMap<String, Py<PyAny>> = HashMap::new();
	65	+ for (k, v) in inputs.iter() {
	66	+ if let Ok(key_str) = k.extract::<String>() {
	67	+ state.insert(key_str, v.into());
	68	+ }
	69	+ }
	70	+
	71	+ let state = Arc::new(tokio::sync::RwLock::new(state));
	72	+ let notify = Arc::new(Notify::new());
	73	+ let cancel = CancellationToken::new();
	74	+
	75	+ // 1. Get asyncio loop and create a future
	76	+ let asyncio = py.import("asyncio")?;
	77	+ let loop_obj = asyncio.call_method0("get_event_loop")?;
	78	+ let future_obj = loop_obj.call_method0("create_future")?;
	79	+
	80	+ let loop_py = loop_obj.into_any().unbind();
	81	+ let fut_py = future_obj.clone().unbind();
	82	+
	83	+ // 2. Decide how to run it based on whether we are in a tokio context
	84	+ let handle = tokio::runtime::Handle::try_current();
	85	+ match handle {
	86	+ Ok(rt) => {
	87	+ rt.spawn(run_dag_internal(
	88	+ nodes, state, notify, cancel, loop_py, fut_py,
	89	+ ));
	90	+ }
	91	+ Err(_) => {
	92	+ std::thread::spawn(move \|\| {
	93	+ let rt = tokio::runtime::Runtime::new().unwrap();
	94	+ rt.block_on(run_dag_internal(
	95	+ nodes, state, notify, cancel, loop_py, fut_py,
	96	+ ));
	97	+ });
	98	+ }
	99	+ }
	100	+
	101	+ Ok(future_obj)
	102	+ }
	103	+}
	104	+
	105	+async fn run_dag_internal(
	106	+ nodes: HashMap<String, DagNode>,
	107	+ state: Arc<tokio::sync::RwLock<HashMap<String, Py<PyAny>>>>,
	108	+ notify: Arc<Notify>,
	109	+ cancel: CancellationToken,
	110	+ loop_py: Py<PyAny>,
	111	+ fut_py: Py<PyAny>,
	112	+) {
	113	+ let mut pending: HashSet<String> = nodes.keys().cloned().collect();
	114	+ let mut completed: HashSet<String> = HashSet::new();
	115	+ let mut running: HashSet<String> = HashSet::new();
	116	+ let mut tasks = tokio::task::JoinSet::new();
	117	+
	118	+ let result = loop {
	119	+ if pending.is_empty() && running.is_empty() {
	120	+ break Ok(());
	121	+ }
	122	+
	123	+ let state_guard = state.read().await;
	124	+ let mut ready = Vec::new();
	125	+ for node_name in &pending {
	126	+ let node = &nodes[node_name];
	127	+ let can_run = node
	128	+ .dependencies
	129	+ .iter()
	130	+ .all(\|d\| state_guard.contains_key(d));
	131	+ if can_run {
	132	+ ready.push(node_name.clone());
	133	+ }
	134	+ }
	135	+ drop(state_guard);
	136	+
	137	+ for r in ready {
	138	+ pending.remove(&r);
	139	+ running.insert(r.clone());
	140	+
	141	+ let node = nodes[&r].clone();
	142	+ let state_clone = state.clone();
	143	+ let notify_clone = notify.clone();
	144	+
	145	+ tasks.spawn_blocking(move \|\| {
	146	+ let res = Python::attach(\|py\| -> Result<Py<PyAny>, String> {
	147	+ let sg = state_clone.blocking_read();
	148	+ let mut args = Vec::new();
	149	+ for dep in &node.dependencies {
	150	+ if let Some(val) = sg.get(dep) {
	151	+ args.push(val.clone_ref(py));
	152	+ }
	153	+ }
	154	+ drop(sg);
	155	+
	156	+ let py_tuple = PyTuple::new(py, args).expect("tuple");
	157	+ match node.handler.call1(py, py_tuple) {
	158	+ Ok(val) => Ok(val),
	159	+ Err(e) => Err(e.to_string()),
	160	+ }
	161	+ });
	162	+ notify_clone.notify_waiters();
	163	+ (r, res)
	164	+ });
	165	+ }
	166	+
	167	+ if !running.is_empty() {
	168	+ tokio::select! {
	169	+ Some(res) = tasks.join_next() => {
	170	+ match res {
	171	+ Ok((name, result)) => {
	172	+ running.remove(&name);
	173	+ completed.insert(name.clone());
	174	+ match result {
	175	+ Ok(py_obj) => {
	176	+ state.write().await.insert(name, py_obj);
	177	+ notify.notify_waiters();
	178	+ }
	179	+ Err(e) => {
	180	+ tracing::error!("DAG node {} failed: {}", name, e);
	181	+ cancel.cancel();
	182	+ break Err(format!("Node {} failed: {}", name, e));
	183	+ }
	184	+ }
	185	+ }
	186	+ Err(e) => {
	187	+ tracing::error!("DAG task join error: {}", e);
	188	+ cancel.cancel();
	189	+ break Err(format!("Join error: {}", e));
	190	+ }
	191	+ }
	192	+ }
	193	+ _ = cancel.cancelled() => {
	194	+ break Err("DAG cancelled".to_string());
	195	+ }
	196	+ }
	197	+ } else if !pending.is_empty() {
	198	+ break Err(format!(
	199	+ "DAG deadlock: nodes {:?} are waiting for dependencies that never arrive",
	200	+ pending
	201	+ ));
	202	+ }
	203	+ };
	204	+
	205	+ let final_state = state.read().await;
	206	+ // 3. Resolve the asyncio future in Python
	207	+ Python::attach(\|py\| {
	208	+ let call_soon_ts = loop_py.getattr(py, "call_soon_threadsafe").unwrap();
	209	+ match result {
	210	+ Ok(_) => {
	211	+ let dict = PyDict::new(py);
	212	+ for (k, v) in final_state.iter() {
	213	+ dict.set_item(k, v).unwrap();
	214	+ }
	215	+ let set_result = fut_py.getattr(py, "set_result").unwrap();
	216	+ if let Err(e) = call_soon_ts.call1(py, (set_result, dict)) {
	217	+ tracing::error!("Failed to call set_result: {:?}", e);
	218	+ e.print(py);
	219	+ }
	220	+ }
	221	+ Err(e) => {
	222	+ let set_exception = fut_py.getattr(py, "set_exception").unwrap();
	223	+ let exceptions = py.import("builtins").unwrap();
	224	+ let runtime_error = exceptions.getattr("RuntimeError").unwrap();
	225	+ let err_obj = runtime_error.call1((e,)).unwrap();
	226	+ if let Err(e) = call_soon_ts.call1(py, (set_exception, err_obj)) {
	227	+ tracing::error!("Failed to call set_exception: {:?}", e);
	228	+ e.print(py);
	229	+ }
	230	+ }
	231	+ }
	232	+ });
	233	+}
0	234

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

+59-0

@@ -0,0 +1,59 @@
	1	+//! Python Database class — wraps PoolManager for JustAPIApp.
	2	+
	3	+use pyo3::prelude::*;
	4	+
	5	+use justapi_core::db::DatabaseConfig;
	6	+
	7	+/// A database connection pool configuration for JustAPI.
	8	+///
	9	+/// Initialized lazily when the app starts via `JustAPIApp.run()`.
	10	+///
	11	+/// Usage:
	12	+/// ```python
	13	+/// from justapi import JustAPIApp, Database
	14	+///
	15	+/// app = JustAPIApp()
	16	+/// app.database = Database("postgres://user:pass@localhost/mydb")
	17	+/// app.run("127.0.0.1:8080")
	18	+/// ```
	19	+#[pyclass(from_py_object)]
	20	+#[derive(Clone)]
	21	+pub struct Database {
	22	+ #[pyo3(get)]
	23	+ pub url: String,
	24	+ #[pyo3(get)]
	25	+ pub max_connections: u32,
	26	+}
	27	+
	28	+#[pymethods]
	29	+impl Database {
	30	+ #[new]
	31	+ #[pyo3(signature = (url, max_connections=10))]
	32	+ fn py_new(url: String, max_connections: u32) -> Self {
	33	+ Self {
	34	+ url,
	35	+ max_connections,
	36	+ }
	37	+ }
	38	+
	39	+ fn __repr__(&self) -> String {
	40	+ format!("Database(url={})", self.url)
	41	+ }
	42	+}
	43	+
	44	+impl Database {
	45	+ pub fn new(url: String, max_connections: u32) -> Self {
	46	+ Self {
	47	+ url,
	48	+ max_connections,
	49	+ }
	50	+ }
	51	+
	52	+ pub fn to_config(&self) -> DatabaseConfig {
	53	+ DatabaseConfig {
	54	+ url: self.url.clone(),
	55	+ max_connections: self.max_connections,
	56	+ kind: None,
	57	+ }
	58	+ }
	59	+}
0	60

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

+61-0

@@ -0,0 +1,61 @@
	1	+use pyo3::prelude::*;
	2	+
	3	+pub mod buffer_test;
	4	+mod dag;
	5	+mod database;
	6	+mod multipart;
	7	+mod native;
	8	+mod rate_limit;
	9	+mod request;
	10	+mod test_client;
	11	+mod websocket;
	12	+
	13	+pub use dag::{Dag, DagNode};
	14	+pub use database::Database;
	15	+pub use multipart::UploadFile;
	16	+pub use native::TokenStreamResponse;
	17	+pub use rate_limit::PyRateLimitResult;
	18	+pub use request::Request;
	19	+pub use websocket::WebSocket;
	20	+
	21	+/// Start the server with hardcoded Rust routes only (no Python app).
	22	+#[pyfunction]
	23	+fn serve(addr: String) -> PyResult<()> {
	24	+ let addr: std::net::SocketAddr = addr
	25	+ .parse()
	26	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(format!("Invalid address: {}", e)))?;
	27	+ let rt = tokio::runtime::Runtime::new()
	28	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(format!("Runtime error: {}", e)))?;
	29	+ rt.block_on(async {
	30	+ let server = justapi_core::Server::new(addr);
	31	+ server
	32	+ .run()
	33	+ .await
	34	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(format!("Server error: {}", e)))
	35	+ })
	36	+}
	37	+
	38	+#[pyfunction]
	39	+fn _test_zero_copy(data: &[u8]) -> buffer_test::ZeroCopyBuffer {
	40	+ buffer_test::ZeroCopyBuffer::new(data.to_vec())
	41	+}
	42	+
	43	+#[pymodule]
	44	+fn _justapi(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
	45	+ m.add_function(wrap_pyfunction!(serve, m)?)?;
	46	+ m.add_class::<buffer_test::ZeroCopyBuffer>()?;
	47	+
	48	+ rate_limit::register(m)?;
	49	+ m.add_function(wrap_pyfunction!(_test_zero_copy, m)?)?;
	50	+ m.add_class::<native::JustAPIApp>()?;
	51	+ m.add_class::<native::TokenStreamResponse>()?;
	52	+ m.add_class::<native::StreamSender>()?;
	53	+ m.add_class::<test_client::JustAPITestClient>()?;
	54	+ m.add_class::<WebSocket>()?;
	55	+ m.add_class::<Request>()?;
	56	+ m.add_class::<UploadFile>()?;
	57	+ m.add_class::<Database>()?;
	58	+ m.add_class::<Dag>()?;
	59	+ m.add_class::<DagNode>()?;
	60	+ Ok(())
	61	+}
0	62

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

+38-0

@@ -0,0 +1,38 @@
	1	+use pyo3::prelude::*;
	2	+use std::fs;
	3	+use std::path::PathBuf;
	4	+
	5	+/// Python-facing `UploadFile` class.
	6	+#[pyclass(name = "UploadFile")]
	7	+pub struct UploadFile {
	8	+ #[pyo3(get)]
	9	+ pub filename: String,
	10	+ #[pyo3(get)]
	11	+ pub content_type: String,
	12	+
	13	+ // Store the path to the temp file
	14	+ temp_path: PathBuf,
	15	+}
	16	+
	17	+#[pymethods]
	18	+impl UploadFile {
	19	+ pub fn read(&self) -> PyResult<Vec<u8>> {
	20	+ let bytes = fs::read(&self.temp_path)?;
	21	+ Ok(bytes)
	22	+ }
	23	+
	24	+ pub fn close(&self) -> PyResult<()> {
	25	+ let _ = fs::remove_file(&self.temp_path);
	26	+ Ok(())
	27	+ }
	28	+}
	29	+
	30	+impl UploadFile {
	31	+ pub fn new(filename: Option<String>, content_type: Option<String>, temp_path: PathBuf) -> Self {
	32	+ Self {
	33	+ filename: filename.unwrap_or_default(),
	34	+ content_type: content_type.unwrap_or_default(),
	35	+ temp_path,
	36	+ }
	37	+ }
	38	+}
0	39

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

+663-0

@@ -0,0 +1,663 @@
	1	+use std::sync::Arc;
	2	+use hyper::Method;
	3	+
	4	+use pyo3::prelude::*;
	5	+use tokio_util::sync::CancellationToken;
	6	+
	7	+use crate::websocket::{WebSocket, WsMessage};
	8	+use futures::{SinkExt, StreamExt};
	9	+
	10	+use justapi_core::router::Router;
	11	+
	12	+use crate::database::Database;
	13	+
	14	+use super::types::*;
	15	+use super::handlers::*;
	16	+
	17	+#[pyclass(name = "_JustAPIApp")]
	18	+pub struct JustAPIApp {
	19	+ pub router: Router<usize>,
	20	+ pub handlers: Vec<Py<PyAny>>,
	21	+ pub schemas: Vec<Option<Py<PyAny>>>,
	22	+ pub schema_jsons: Vec<Option<String>>,
	23	+ pub batch_configs: Vec<Option<(usize, u64)>>,
	24	+ pub plugins: Vec<Py<PyAny>>,
	25	+ pub database: Option<Database>,
	26	+ pub grpc_addr: Option<std::net::SocketAddr>,
	27	+ pub grpc_handlers: std::collections::HashMap<String, Py<PyAny>>,
	28	+ pub ws_routes: std::collections::HashMap<String, Py<PyAny>>,
	29	+ pub wasm_bytes: Option<Vec<u8>>,
	30	+ pub gateway_config: Option<String>,
	31	+ pub circuit_breaker_config: Option<(usize, u64)>,
	32	+ pub coalesce_headers: Option<Vec<String>>,
	33	+}
	34	+
	35	+#[pymethods]
	36	+impl JustAPIApp {
	37	+ #[new]
	38	+ fn new() -> Self {
	39	+ Self {
	40	+ router: Router::new(),
	41	+ handlers: Vec::new(),
	42	+ schemas: Vec::new(),
	43	+ schema_jsons: Vec::new(),
	44	+ batch_configs: Vec::new(),
	45	+ plugins: Vec::new(),
	46	+ database: None,
	47	+ grpc_addr: None,
	48	+ grpc_handlers: std::collections::HashMap::new(),
	49	+ ws_routes: std::collections::HashMap::new(),
	50	+ wasm_bytes: None,
	51	+ gateway_config: None,
	52	+ circuit_breaker_config: None,
	53	+ coalesce_headers: None,
	54	+ }
	55	+ }
	56	+
	57	+ #[pyo3(name = "enable_gateway")]
	58	+ fn enable_gateway(&mut self, path: &str) {
	59	+ self.gateway_config = Some(path.to_string());
	60	+ }
	61	+
	62	+ #[pyo3(name = "enable_circuit_breaker")]
	63	+ fn enable_circuit_breaker(&mut self, failure_threshold: usize, reset_timeout_ms: u64) {
	64	+ self.circuit_breaker_config = Some((failure_threshold, reset_timeout_ms));
	65	+ }
	66	+
	67	+ /// Enable request coalescing (singleflight). When many concurrent, identical
	68	+ /// requests hit the same route, only one reaches the handler; the rest share
	69	+ /// its response. `headers`, if given, are included in the coalesce key
	70	+ /// (e.g. `["accept"]` so distinct representations are not collapsed).
	71	+ #[pyo3(name = "enable_request_coalescing")]
	72	+ fn enable_request_coalescing(&mut self, headers: Option<Vec<String>>) {
	73	+ self.coalesce_headers = Some(headers.unwrap_or_default());
	74	+ }
	75	+
	76	+ fn load_wasm_middleware(&mut self, path: &str) -> PyResult<()> {
	77	+ let bytes = std::fs::read(path).map_err(\|e\| {
	78	+ pyo3::exceptions::PyIOError::new_err(format!(
	79	+ "Failed to read WASM file {}: {}",
	80	+ path, e
	81	+ ))
	82	+ })?;
	83	+ self.wasm_bytes = Some(bytes);
	84	+ Ok(())
	85	+ }
	86	+
	87	+ /// Register a plugin. This immediately calls `plugin.build(self)`.
	88	+ #[pyo3(name = "use")]
	89	+ fn use_plugin(slf: Py<Self>, py: Python<'_>, plugin: Py<PyAny>) -> PyResult<()> {
	90	+ plugin.call_method1(py, "build", (slf.clone_ref(py),))?;
	91	+ slf.borrow_mut(py).plugins.push(plugin);
	92	+ Ok(())
	93	+ }
	94	+
	95	+ fn get(&mut self, path: &str, handler: Py<PyAny>) -> PyResult<()> {
	96	+ let id = self.handlers.len();
	97	+ self.handlers.push(handler);
	98	+ self.schemas.push(None);
	99	+ self.schema_jsons.push(None);
	100	+ self.batch_configs.push(None);
	101	+ self.router
	102	+ .insert(Method::GET, path, id)
	103	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	104	+ }
	105	+
	106	+ #[allow(clippy::too_many_arguments)]
	107	+ #[pyo3(signature = (path, handler, body_schema=None, schema=None, batch_size=None, batch_window_ms=None))]
	108	+ fn post(
	109	+ &mut self,
	110	+ py: Python<'_>,
	111	+ path: &str,
	112	+ handler: Py<PyAny>,
	113	+ body_schema: Option<Py<PyAny>>,
	114	+ schema: Option<Py<PyAny>>,
	115	+ batch_size: Option<usize>,
	116	+ batch_window_ms: Option<u64>,
	117	+ ) -> PyResult<()> {
	118	+ let id = self.handlers.len();
	119	+ self.handlers.push(handler);
	120	+ self.schemas.push(body_schema);
	121	+ self.schema_jsons.push(resolve_schema_json(py, schema)?);
	122	+ self.batch_configs
	123	+ .push(batch_size.map(\|s\| (s, batch_window_ms.unwrap_or(10))));
	124	+ self.router
	125	+ .insert(Method::POST, path, id)
	126	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	127	+ }
	128	+
	129	+ #[allow(clippy::too_many_arguments)]
	130	+ #[pyo3(signature = (path, handler, body_schema=None, schema=None, batch_size=None, batch_window_ms=None))]
	131	+ fn put(
	132	+ &mut self,
	133	+ py: Python<'_>,
	134	+ path: &str,
	135	+ handler: Py<PyAny>,
	136	+ body_schema: Option<Py<PyAny>>,
	137	+ schema: Option<Py<PyAny>>,
	138	+ batch_size: Option<usize>,
	139	+ batch_window_ms: Option<u64>,
	140	+ ) -> PyResult<()> {
	141	+ let id = self.handlers.len();
	142	+ self.handlers.push(handler);
	143	+ self.schemas.push(body_schema);
	144	+ self.schema_jsons.push(resolve_schema_json(py, schema)?);
	145	+ self.batch_configs
	146	+ .push(batch_size.map(\|s\| (s, batch_window_ms.unwrap_or(10))));
	147	+ self.router
	148	+ .insert(Method::PUT, path, id)
	149	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	150	+ }
	151	+
	152	+ #[allow(clippy::too_many_arguments)]
	153	+ #[pyo3(signature = (path, handler, body_schema=None, schema=None, batch_size=None, batch_window_ms=None))]
	154	+ fn patch(
	155	+ &mut self,
	156	+ py: Python<'_>,
	157	+ path: &str,
	158	+ handler: Py<PyAny>,
	159	+ body_schema: Option<Py<PyAny>>,
	160	+ schema: Option<Py<PyAny>>,
	161	+ batch_size: Option<usize>,
	162	+ batch_window_ms: Option<u64>,
	163	+ ) -> PyResult<()> {
	164	+ let id = self.handlers.len();
	165	+ self.handlers.push(handler);
	166	+ self.schemas.push(body_schema);
	167	+ self.schema_jsons.push(resolve_schema_json(py, schema)?);
	168	+ self.batch_configs
	169	+ .push(batch_size.map(\|s\| (s, batch_window_ms.unwrap_or(10))));
	170	+ self.router
	171	+ .insert(Method::PATCH, path, id)
	172	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	173	+ }
	174	+
	175	+ fn delete(&mut self, path: &str, handler: Py<PyAny>) -> PyResult<()> {
	176	+ let id = self.handlers.len();
	177	+ self.handlers.push(handler);
	178	+ self.schemas.push(None);
	179	+ self.schema_jsons.push(None);
	180	+ self.batch_configs.push(None);
	181	+ self.router
	182	+ .insert(Method::DELETE, path, id)
	183	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	184	+ }
	185	+
	186	+ /// Register a route for the HTTP QUERY method (RFC 10008): safe,
	187	+ /// idempotent, and body-carrying (like POST). The `experimental` flag
	188	+ /// is reserved for tagging the operation in generated OpenAPI specs.
	189	+ #[pyo3(signature = (path, handler, body_schema=None, schema=None, experimental=None))]
	190	+ fn query(
	191	+ &mut self,
	192	+ py: Python<'_>,
	193	+ path: &str,
	194	+ handler: Py<PyAny>,
	195	+ body_schema: Option<Py<PyAny>>,
	196	+ schema: Option<Py<PyAny>>,
	197	+ experimental: Option<bool>,
	198	+ ) -> PyResult<()> {
	199	+ let _ = experimental;
	200	+ let id = self.handlers.len();
	201	+ self.handlers.push(handler);
	202	+ self.schemas.push(body_schema);
	203	+ self.schema_jsons.push(resolve_schema_json(py, schema)?);
	204	+ self.batch_configs.push(None);
	205	+ self.router
	206	+ .insert(justapi_core::query_method(), path, id)
	207	+ .map_err(\|e\| pyo3::exceptions::PyValueError::new_err(e.to_string()))
	208	+ }
	209	+
	210	+ /// Set the database configuration.
	211	+ #[pyo3(signature = (db))]
	212	+ fn set_database(&mut self, _py: Python<'_>, db: &Bound<'_, PyAny>) -> PyResult<()> {
	213	+ if let Ok(db_obj) = db.extract::<Database>() {
	214	+ self.database = Some(db_obj);
	215	+ Ok(())
	216	+ } else if let Ok(url) = db.extract::<String>() {
	217	+ self.database = Some(Database::new(url, 10));
	218	+ Ok(())
	219	+ } else {
	220	+ Err(pyo3::exceptions::PyTypeError::new_err(
	221	+ "database must be a Database object or a connection string",
	222	+ ))
	223	+ }
	224	+ }
	225	+
	226	+ /// Get the database configuration, or None.
	227	+ fn get_database(&self) -> Option<Database> {
	228	+ self.database.clone()
	229	+ }
	230	+
	231	+ fn set_grpc_addr(&mut self, addr: &str) -> PyResult<()> {
	232	+ self.grpc_addr = Some(addr.parse().map_err(\|e\| {
	233	+ pyo3::exceptions::PyValueError::new_err(format!("Invalid gRPC address: {}", e))
	234	+ })?);
	235	+ Ok(())
	236	+ }
	237	+
	238	+ fn add_grpc_service(&mut self, path: String, handler: Py<PyAny>) {
	239	+ self.grpc_handlers.insert(path, handler);
	240	+ }
	241	+
	242	+ /// Register a WebSocket handler for `path`. The handler is an async Python
	243	+ /// function receiving a single `WebSocket` argument.
	244	+ fn websocket(&mut self, path: String, handler: Py<PyAny>) -> PyResult<()> {
	245	+ if self.ws_routes.contains_key(&path) {
	246	+ return Err(pyo3::exceptions::PyValueError::new_err(format!(
	247	+ "WebSocket route already registered: {}",
	248	+ path
	249	+ )));
	250	+ }
	251	+ self.ws_routes.insert(path, handler);
	252	+ Ok(())
	253	+ }
	254	+
	255	+ /// Start the server and begin accepting requests.
	256	+ fn run(&mut self, py: Python<'_>, addr: &str) -> PyResult<()> {
	257	+ let addr: std::net::SocketAddr = addr.parse().map_err(\|e\| {
	258	+ pyo3::exceptions::PyValueError::new_err(format!("Invalid address: {}", e))
	259	+ })?;
	260	+
	261	+ // Build OpenAPI spec from registered routes before taking ownership.
	262	+ let openapi_spec = {
	263	+ use justapi_core::openapi::*;
	264	+ let routes: Vec<(hyper::Method, String)> = self.router.list_routes().to_vec();
	265	+ let mut registry = OpenApiRegistry::new();
	266	+ for (method, path) in &routes {
	267	+ let _path_params: Vec<String> = path
	268	+ .split('/')
	269	+ .filter(\|s\| s.starts_with('{') && s.ends_with('}'))
	270	+ .map(\|s\| s[1..s.len() - 1].to_string())
	271	+ .collect();
	272	+ let is_body_method = matches!(
	273	+ *method,
	274	+ hyper::Method::POST \| hyper::Method::PUT \| hyper::Method::PATCH
	275	+ );
	276	+ registry.register(RouteMeta {
	277	+ method: method.clone(),
	278	+ path: path.clone(),
	279	+ summary: None,
	280	+ description: None,
	281	+ tags: vec![],
	282	+ request_body_schema: if is_body_method {
	283	+ Some(serde_json::json!({}))
	284	+ } else {
	285	+ None
	286	+ },
	287	+ response_schema: Some(serde_json::json!({})),
	288	+ deprecated: false,
	289	+ experimental: false,
	290	+ });
	291	+ }
	292	+ if routes.is_empty() {
	293	+ None
	294	+ } else {
	295	+ Some(
	296	+ serde_json::to_string_pretty(&registry.generate("JustAPI", "2.0.0"))
	297	+ .unwrap_or_default(),
	298	+ )
	299	+ }
	300	+ };
	301	+
	302	+ let router = Arc::new(std::mem::take(&mut self.router));
	303	+ let handlers = Arc::new(std::mem::take(&mut self.handlers));
	304	+ let schemas = Arc::new(std::mem::take(&mut self.schemas));
	305	+ let schema_jsons = Arc::new(std::mem::take(&mut self.schema_jsons));
	306	+ let batch_configs_arc = Arc::new(std::mem::take(&mut self.batch_configs));
	307	+ let database_config = self.database.take().map(\|d\| d.to_config());
	308	+ let plugins = std::mem::take(&mut self.plugins);
	309	+ let grpc_addr = self.grpc_addr.take();
	310	+ let grpc_handlers = Arc::new(std::mem::take(&mut self.grpc_handlers));
	311	+ let ws_routes = std::mem::take(&mut self.ws_routes);
	312	+ let wasm_bytes = self.wasm_bytes.take();
	313	+ let gateway_config_path = self.gateway_config.take();
	314	+ let cb_config = self.circuit_breaker_config.take();
	315	+ let coalesce_headers = self.coalesce_headers.take();
	316	+
	317	+ let shutdown = CancellationToken::new();
	318	+ let shutdown_signal = shutdown.clone();
	319	+
	320	+ for plugin in &plugins {
	321	+ call_plugin_hook(py, plugin, "on_startup")?;
	322	+ }
	323	+
	324	+ // `py.detach` blocks the calling thread while releasing the GIL
	325	+ // (standard server-entrypoint behavior, like `uvicorn.run`), keeping
	326	+ // the process alive until shutdown. Handler threads call back into
	327	+ // Python via `Python::attach`.
	328	+ let result = py.detach(move \|\| -> Result<(), anyhow::Error> {
	329	+ let rt = tokio::runtime::Runtime::new()?;
	330	+ rt.block_on(async {
	331	+ // Initialize database pool if configured, and build transaction-aware handler.
	332	+ let db_pool = if let Some(ref config) = database_config {
	333	+ let mut mgr = justapi_core::db::PoolManager::new();
	334	+ match mgr.init("", config.clone()).await {
	335	+ Ok(pool) => {
	336	+ tracing::info!("Database pool initialized ({:?})", pool.kind());
	337	+ Some(pool)
	338	+ }
	339	+ Err(e) => {
	340	+ tracing::error!("Failed to initialize database pool: {}", e);
	341	+ return Err(anyhow::anyhow!("Database init error: {}", e));
	342	+ }
	343	+ }
	344	+ } else {
	345	+ None
	346	+ };
	347	+
	348	+ let db_url_str = database_config.as_ref().map(\|c\| c.url.clone());
	349	+
	350	+ let mut batchers = Vec::new();
	351	+ for (id, config) in batch_configs_arc.iter().enumerate() {
	352	+ if let Some((max_size, window_ms)) = config {
	353	+ let handlers_arc = handlers.clone();
	354	+ let db_url_str2 = db_url_str.clone();
	355	+ let batcher = justapi_core::batching::start_batcher(
	356	+ *max_size,
	357	+ std::time::Duration::from_millis(*window_ms),
	358	+ move \|batch: Vec<BatchedReq>\| {
	359	+ let handlers_arc = handlers_arc.clone();
	360	+ let db_url_str2 = db_url_str2.clone();
	361	+ async move {
	362	+ tokio::task::spawn_blocking(move \|\| {
	363	+ Python::attach(\|py\| {
	364	+ let py_handler = handlers_arc[id].clone_ref(py);
	365	+ let helper = get_helper(py);
	366	+ let py_list = pyo3::types::PyList::empty(py);
	367	+ for req in &batch {
	368	+ let r = Bound::new(py, crate::request::Request::new(
	369	+ py,
	370	+ req.method.clone(),
	371	+ req.path.clone(),
	372	+ req.path_params.to_vec(),
	373	+ req.query_string.to_vec(),
	374	+ req.headers.to_vec(),
	375	+ req.body.to_vec(),
	376	+ db_url_str2.as_deref().map(\|s\| s.to_string()),
	377	+ None,
	378	+ )).unwrap();
	379	+ py_list.append(r).unwrap();
	380	+ }
	381	+
	382	+ let result = helper.call_batch_handler.bind(py).call1((py_handler.bind(py), py_list));
	383	+ let final_res = match result {
	384	+ Ok(res) => {
	385	+ let is_future = res.hasattr("result").unwrap_or(false) && !res.is_instance_of::<pyo3::types::PyList>();
	386	+ if is_future {
	387	+ res.call_method0("result")
	388	+ } else {
	389	+ Ok(res)
	390	+ }
	391	+ }
	392	+ Err(e) => Err(e),
	393	+ };
	394	+
	395	+ let mut results = Vec::new();
	396	+ match final_res {
	397	+ Ok(res_list) => {
	398	+ if let Ok(list) = res_list.extract::<Vec<pyo3::Bound<'_, pyo3::PyAny>>>() {
	399	+ for item in list.iter() {
	400	+ let status: u16 = item.get_item("status").ok().and_then(\|v\| v.extract().ok()).unwrap_or(200);
	401	+ let headers: Vec<(Vec<u8>, Vec<u8>)> = item.get_item("headers").ok().and_then(\|v\| v.extract().ok()).unwrap_or_default();
	402	+ let body: Vec<u8> = item.get_item("body").ok().and_then(\|v\| v.extract().ok()).unwrap_or_default();
	403	+ results.push(NativeResponse { status, headers, body: NativeBody::Bytes(body) });
	404	+ }
	405	+ }
	406	+ }
	407	+ Err(e) => {
	408	+ tracing::error!("Batch handler error: {}", e);
	409	+ for _ in 0..batch.len() {
	410	+ results.push(NativeResponse {
	411	+ status: 500,
	412	+ headers: vec![],
	413	+ body: NativeBody::Bytes(b"Internal Server Error".to_vec()),
	414	+ });
	415	+ }
	416	+ }
	417	+ }
	418	+
	419	+ // Handle case where user returns wrong number of elements
	420	+ while results.len() < batch.len() {
	421	+ results.push(NativeResponse {
	422	+ status: 500,
	423	+ headers: vec![],
	424	+ body: NativeBody::Bytes(b"Batch length mismatch".to_vec()),
	425	+ });
	426	+ }
	427	+ results.truncate(batch.len());
	428	+
	429	+ results
	430	+ })
	431	+ }).await.unwrap_or_default()
	432	+ }
	433	+ }
	434	+ );
	435	+ batchers.push(Some(batcher));
	436	+ } else {
	437	+ batchers.push(None);
	438	+ }
	439	+ }
	440	+ let batchers = Arc::new(batchers);
	441	+ let handler = make_native_handler(
	442	+ router,
	443	+ handlers,
	444	+ schemas,
	445	+ schema_jsons,
	446	+ batchers,
	447	+ db_pool.clone(),
	448	+ db_url_str,
	449	+ );
	450	+
	451	+ // Install Ctrl+C signal handler to trigger graceful shutdown.
	452	+ let signal_token = shutdown.clone();
	453	+ tokio::spawn(async move {
	454	+ tokio::signal::ctrl_c().await.ok();
	455	+ tracing::info!("Ctrl+C received, initiating graceful shutdown");
	456	+ signal_token.cancel();
	457	+ });
	458	+
	459	+ let mut server = justapi_core::Server::new(addr)
	460	+ .with_handler(handler)
	461	+ .with_shutdown(shutdown_signal);
	462	+ if let Some(ref spec) = openapi_spec {
	463	+ server = server.with_openapi_spec(spec.clone());
	464	+ }
	465	+
	466	+ if let Some(ref wasm_bytes) = wasm_bytes {
	467	+ server = match server.with_wasm_middleware(wasm_bytes) {
	468	+ Ok(s) => s,
	469	+ Err(e) => {
	470	+ tracing::error!("Failed to initialize WASM middleware: {}", e);
	471	+ return Err(anyhow::anyhow!("WASM middleware error: {}", e));
	472	+ }
	473	+ };
	474	+ }
	475	+
	476	+ if let Some((failure_threshold, reset_timeout_ms)) = cb_config {
	477	+ tracing::info!("Enabling Global Circuit Breaker (threshold: {}, reset: {}ms)", failure_threshold, reset_timeout_ms);
	478	+ let config = justapi_core::resilience::CircuitBreakerConfig {
	479	+ failure_threshold: failure_threshold as u32,
	480	+ open_timeout: std::time::Duration::from_millis(reset_timeout_ms),
	481	+ ..Default::default()
	482	+ };
	483	+ server = server.add_circuit_breaker(config);
	484	+ }
	485	+
	486	+ if let Some(headers) = coalesce_headers {
	487	+ let mut coalescer = justapi_core::coalesce::RequestCoalescer::new();
	488	+ if !headers.is_empty() {
	489	+ let mut names = Vec::new();
	490	+ for h in &headers {
	491	+ match hyper::header::HeaderName::from_bytes(h.as_bytes()) {
	492	+ Ok(n) => names.push(n),
	493	+ Err(e) => {
	494	+ tracing::error!(
	495	+ "Invalid coalesce header {} ignored: {}",
	496	+ h,
	497	+ e
	498	+ );
	499	+ }
	500	+ }
	501	+ }
	502	+ coalescer = coalescer.with_headers(&names);
	503	+ }
	504	+ tracing::info!(
	505	+ "Enabling request coalescing (singleflight){}",
	506	+ if headers.is_empty() {
	507	+ String::new()
	508	+ } else {
	509	+ format!(" with key headers: {:?}", headers)
	510	+ }
	511	+ );
	512	+ server = server.add_middleware(coalescer);
	513	+ }
	514	+
	515	+ if let Some(path) = gateway_config_path {
	516	+ tracing::info!("Starting Gateway with hot reloading for config: {}", path);
	517	+ let gateway_state = justapi_core::gateway::GatewayState::new(&path);
	518	+ if let Err(e) = gateway_state.clone().watch() {
	519	+ tracing::error!("Failed to start gateway config file watcher: {}", e);
	520	+ }
	521	+ server = server.add_gateway(gateway_state);
	522	+ }
	523	+
	524	+ if let (Some(gaddr), true) = (grpc_addr, !grpc_handlers.is_empty()) {
	525	+ let g_handlers = grpc_handlers.clone();
	526	+ let grpc_h = Box::new(move \|uri: http::Uri, body: Vec<u8>\| {
	527	+ let g_handlers = g_handlers.clone();
	528	+ Box::pin(async move {
	529	+ let path = uri.path().to_string();
	530	+ let (tx, rx) = tokio::sync::oneshot::channel();
	531	+ tokio::task::spawn_blocking(move \|\| {
	532	+ Python::attach(\|py\| {
	533	+ if let Some(h) = g_handlers.get(&path) {
	534	+ let py_body = pyo3::types::PyBytes::new(py, &body);
	535	+ match h.call1(py, (py_body,)) {
	536	+ Ok(res) => {
	537	+ if let Ok(b) = res.extract::<Vec<u8>>(py) {
	538	+ let _ = tx.send(Ok(b));
	539	+ } else {
	540	+ let _ = tx.send(Err(tonic::Status::internal("Invalid return type from python gRPC handler")));
	541	+ }
	542	+ }
	543	+ Err(e) => {
	544	+ let err_msg = e.to_string();
	545	+ let _ = tx.send(Err(tonic::Status::internal(err_msg)));
	546	+ }
	547	+ }
	548	+ } else {
	549	+ let _ = tx.send(Err(tonic::Status::unimplemented("Method not found")));
	550	+ }
	551	+ });
	552	+ });
	553	+ rx.await.unwrap_or_else(\|_\| Err(tonic::Status::internal("Channel dropped")))
	554	+ }) as std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, tonic::Status>> + Send + 'static>>
	555	+ });
	556	+ server = server.with_grpc(gaddr, grpc_h);
	557	+ }
	558	+
	559	+ // WebSocket handlers: build a WsHandler that bridges the
	560	+ // accepted tokio-tungstenite stream to a Python handler.
	561	+ if !ws_routes.is_empty() {
	562	+ let ws_routes_arc = Arc::new(ws_routes);
	563	+ let ws_handler: justapi_core::server::WsHandler = Arc::new(move \|path, mut read, mut write\| {
	564	+ let ws_routes = ws_routes_arc.clone();
	565	+ Box::pin(async move {
	566	+ let (out_tx, mut out_rx) =
	567	+ tokio::sync::mpsc::unbounded_channel::<WsMessage>();
	568	+ let (in_tx, in_rx) = tokio::sync::mpsc::unbounded_channel::<WsMessage>();
	569	+ let incoming =
	570	+ std::sync::Arc::new(tokio::sync::Mutex::new(in_rx));
	571	+
	572	+ // Inbound WebSocket messages -> Python incoming channel.
	573	+ tokio::spawn(async move {
	574	+ while let Some(msg) = read.next().await {
	575	+ match msg {
	576	+ Ok(tokio_tungstenite::tungstenite::Message::Text(t)) => {
	577	+ let _ = in_tx
	578	+ .send(WsMessage::Text(t.to_string()));
	579	+ }
	580	+ Ok(tokio_tungstenite::tungstenite::Message::Binary(b)) => {
	581	+ let _ = in_tx.send(WsMessage::Bytes(b.to_vec()));
	582	+ }
	583	+ Ok(tokio_tungstenite::tungstenite::Message::Close(_))
	584	+ \| Err(_) => {
	585	+ let _ = in_tx.send(WsMessage::Close);
	586	+ break;
	587	+ }
	588	+ _ => {}
	589	+ }
	590	+ }
	591	+ });
	592	+
	593	+ // Python outgoing channel -> WebSocket sink.
	594	+ tokio::spawn(async move {
	595	+ while let Some(msg) = out_rx.recv().await {
	596	+ let frame = match &msg {
	597	+ WsMessage::Text(t) => {
	598	+ tokio_tungstenite::tungstenite::Message::Text(t.into())
	599	+ }
	600	+ WsMessage::Bytes(b) => {
	601	+ tokio_tungstenite::tungstenite::Message::Binary(
	602	+ b.clone(),
	603	+ )
	604	+ }
	605	+ WsMessage::Close => {
	606	+ tokio_tungstenite::tungstenite::Message::Close(None)
	607	+ }
	608	+ };
	609	+ if write.send(frame).await.is_err() {
	610	+ break;
	611	+ }
	612	+ if matches!(msg, WsMessage::Close) {
	613	+ break;
	614	+ }
	615	+ }
	616	+ let _ = write.close().await;
	617	+ });
	618	+
	619	+ // Invoke the Python handler on the daemon event loop.
	620	+ Python::attach(\|py\| {
	621	+ if let Some(h) = ws_routes.get(&path) {
	622	+ match h
	623	+ .bind(py)
	624	+ .call1((WebSocket::new(incoming, out_tx),))
	625	+ {
	626	+ Ok(coro) => {
	627	+ let helper = get_helper(py);
	628	+ if let Err(e) =
	629	+ helper.run_ws_handler.bind(py).call1((coro,))
	630	+ {
	631	+ tracing::error!(
	632	+ "Failed to schedule WebSocket handler: {}",
	633	+ e
	634	+ );
	635	+ }
	636	+ }
	637	+ Err(e) => {
	638	+ tracing::error!("WebSocket handler call error: {}", e)
	639	+ }
	640	+ }
	641	+ }
	642	+ });
	643	+
	644	+ Ok(())
	645	+ })
	646	+ });
	647	+ server = server.with_ws(ws_handler);
	648	+ }
	649	+
	650	+ server.run().await
	651	+ })
	652	+ });
	653	+
	654	+ for plugin in &plugins {
	655	+ if let Err(e) = call_plugin_hook(py, plugin, "on_shutdown") {
	656	+ tracing::error!("Error in plugin on_shutdown: {}", e);
	657	+ }
	658	+ }
	659	+
	660	+ result
	661	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(format!("Server error: {}", e)))
	662	+ }
	663	+}
0	664

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

+651-0

@@ -0,0 +1,651 @@
	1	+use std::sync::Arc;
	2	+
	3	+use http_body_util::BodyExt;
	4	+use hyper::body::Bytes;
	5	+use hyper::{Method, Request, Response, StatusCode};
	6	+use pyo3::prelude::*;
	7	+
	8	+use justapi_core::router::Router;
	9	+use justapi_core::{json_response, middleware::HandlerFn, ResponseBody};
	10	+
	11	+use justapi_core::db::AnyPool;
	12	+
	13	+use super::types::*;
	14	+#[allow(clippy::too_many_arguments)]
	15	+pub(crate) fn call_python_handler(
	16	+ py: Python<'_>,
	17	+ handler: &Py<PyAny>,
	18	+ schema: Option<&Py<PyAny>>,
	19	+ path_params: &[(String, String)],
	20	+ query_string: &[u8],
	21	+ headers: &[(Vec<u8>, Vec<u8>)],
	22	+ body: &[u8],
	23	+ db_url: Option<&str>,
	24	+ trace_ctx: Option<(String, String)>,
	25	+ multipart_form: Option<Py<pyo3::types::PyDict>>,
	26	+ method: &str,
	27	+ path: &str,
	28	+) -> NativeResponse {
	29	+ let helper = get_helper(py);
	30	+
	31	+ // Set trace context in Python contextvars for distributed tracing
	32	+ if let Some((trace_id, span_id)) = trace_ctx {
	33	+ let _ = helper.set_trace_context.bind(py).call1((trace_id, span_id));
	34	+ }
	35	+
	36	+ // Validate body if a schema is registered
	37	+ if let Some(schema) = schema {
	38	+ let validation_result = helper.validate_body.bind(py).call1((schema, body));
	39	+ match validation_result {
	40	+ Ok(errors) => {
	41	+ if let Ok(error_list) = errors.extract::<Vec<String>>() {
	42	+ if !error_list.is_empty() {
	43	+ let error_body = serde_json::json!({
	44	+ "type": "https://justapi.dev/errors/validation",
	45	+ "title": "Validation Error",
	46	+ "status": 422,
	47	+ "detail": error_list.join("; "),
	48	+ "errors": error_list.iter().map(\|e\| {
	49	+ serde_json::json!({"message": e})
	50	+ }).collect::<Vec<_>>(),
	51	+ })
	52	+ .to_string();
	53	+ return NativeResponse {
	54	+ status: 422,
	55	+ headers: vec![(
	56	+ b"content-type".to_vec(),
	57	+ b"application/problem+json".to_vec(),
	58	+ )],
	59	+ body: NativeBody::Bytes(error_body.into_bytes()),
	60	+ };
	61	+ }
	62	+ }
	63	+ }
	64	+ Err(e) => {
	65	+ tracing::error!("Body validation error: {}", e);
	66	+ }
	67	+ }
	68	+ }
	69	+
	70	+ let request = Bound::new(
	71	+ py,
	72	+ crate::request::Request::new(
	73	+ py,
	74	+ method.to_string(),
	75	+ path.to_string(),
	76	+ path_params.to_vec(),
	77	+ query_string.to_vec(),
	78	+ headers.to_vec(),
	79	+ body.to_vec(),
	80	+ db_url.map(\|s\| s.to_string()),
	81	+ multipart_form,
	82	+ ),
	83	+ )
	84	+ .unwrap();
	85	+
	86	+ let result = helper.call_handler.bind(py).call1((handler, request));
	87	+
	88	+ let final_res = match result {
	89	+ Ok(res) => {
	90	+ let is_future = res.hasattr("result").unwrap_or(false)
	91	+ && !res.is_instance_of::<pyo3::types::PyDict>();
	92	+ if is_future {
	93	+ res.call_method0("result")
	94	+ } else {
	95	+ Ok(res)
	96	+ }
	97	+ }
	98	+ Err(e) => Err(e),
	99	+ };
	100	+
	101	+ match final_res {
	102	+ Ok(res) => {
	103	+ if let Ok(ts_res) = res.extract::<pyo3::PyRef<'_, TokenStreamResponse>>() {
	104	+ let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
	105	+ let sender = StreamSender { tx };
	106	+ let pump = helper.pump_stream.bind(py);
	107	+ if let Err(e) = pump.call1((ts_res.generator.clone_ref(py), sender)) {
	108	+ tracing::error!("Failed to pump stream: {}", e);
	109	+ }
	110	+ return NativeResponse {
	111	+ status: ts_res.status,
	112	+ headers: ts_res.headers.clone(),
	113	+ body: NativeBody::Stream(rx),
	114	+ };
	115	+ }
	116	+
	117	+ let wrapped_result = helper.wrap_result.bind(py).call1((res,)).unwrap();
	118	+
	119	+ let status: u16 = wrapped_result
	120	+ .get_item("status")
	121	+ .ok()
	122	+ .and_then(\|v\| v.extract::<u16>().ok())
	123	+ .unwrap_or(200);
	124	+ let resp_headers: Vec<(Vec<u8>, Vec<u8>)> = wrapped_result
	125	+ .get_item("headers")
	126	+ .ok()
	127	+ .and_then(\|v\| v.extract::<Vec<(Vec<u8>, Vec<u8>)>>().ok())
	128	+ .unwrap_or_default();
	129	+ let resp_body: Vec<u8> = wrapped_result
	130	+ .get_item("body")
	131	+ .ok()
	132	+ .and_then(\|v\| v.extract::<Vec<u8>>().ok())
	133	+ .unwrap_or_default();
	134	+ NativeResponse {
	135	+ status,
	136	+ headers: resp_headers,
	137	+ body: NativeBody::Bytes(resp_body),
	138	+ }
	139	+ }
	140	+ Err(e) => {
	141	+ tracing::error!("Native handler error: {}", e);
	142	+ NativeResponse {
	143	+ status: 500,
	144	+ headers: vec![],
	145	+ body: NativeBody::Bytes(format!("Internal Server Error: {}", e).into_bytes()),
	146	+ }
	147	+ }
	148	+ }
	149	+}
	150	+
	151	+pub(crate) fn nr_to_response(nr: NativeResponse) -> hyper::Response<ResponseBody> {
	152	+ let mut resp = match nr.body {
	153	+ NativeBody::Bytes(b) => {
	154	+ let mut r = Response::new(ResponseBody::new(
	155	+ http_body_util::Full::new(Bytes::from(b))
	156	+ .map_err(\|e: std::convert::Infallible\| -> anyhow::Error { match e {} }),
	157	+ ));
	158	+ *r.status_mut() =
	159	+ StatusCode::from_u16(nr.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
	160	+ r
	161	+ }
	162	+ NativeBody::Stream(rx) => {
	163	+ let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
	164	+ justapi_core::streaming_response(
	165	+ StatusCode::from_u16(nr.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
	166	+ "text/event-stream",
	167	+ stream,
	168	+ )
	169	+ }
	170	+ };
	171	+
	172	+ for (name, value) in &nr.headers {
	173	+ if let (Ok(n), Ok(v)) = (
	174	+ http::HeaderName::from_bytes(name),
	175	+ http::HeaderValue::from_bytes(value),
	176	+ ) {
	177	+ resp.headers_mut().insert(n, v);
	178	+ }
	179	+ }
	180	+ resp
	181	+}
	182	+
	183	+/// Python-facing native API app — the FastAPI replacement.
	184	+///
	185	+/// Usage:
	186	+/// ```python
	187	+/// from justapi import JustAPIApp, Schema
	188	+///
	189	+/// class UserSchema(Schema):
	190	+/// name: str
	191	+/// email: str
	192	+/// age: int \| None = None
	193	+///
	194	+/// app = JustAPIApp()
	195	+///
	196	+/// app.post("/users", create_user, schema=UserSchema)
	197	+/// app.run("127.0.0.1:8080")
	198	+/// ```
	199	+pub(crate) fn make_native_handler(
	200	+ router: Arc<Router<usize>>,
	201	+ handlers: Arc<Vec<Py<PyAny>>>,
	202	+ schemas: Arc<Vec<Option<Py<PyAny>>>>,
	203	+ schema_jsons: Arc<Vec<Option<String>>>,
	204	+ batchers: Arc<Vec<Option<justapi_core::batching::Batcher<BatchedReq, NativeResponse>>>>,
	205	+ db_pool: Option<AnyPool>,
	206	+ db_url_str: Option<String>,
	207	+) -> HandlerFn {
	208	+ let db_pool = Arc::new(db_pool);
	209	+ let db_url = Arc::new(db_url_str);
	210	+ Arc::new(move \|req: Request<hyper::body::Incoming>\| {
	211	+ let router = router.clone();
	212	+ let handlers = handlers.clone();
	213	+ let schemas = schemas.clone();
	214	+ let schema_jsons = schema_jsons.clone();
	215	+ let batchers = batchers.clone();
	216	+ let db_pool = db_pool.clone();
	217	+ let db_url = db_url.clone();
	218	+ Box::pin(async move {
	219	+ let method = req.method().clone();
	220	+ let path = req.uri().path().to_string();
	221	+ let query_string = req.uri().query().unwrap_or("").as_bytes().to_vec();
	222	+
	223	+ // RFC 10008: QUERY requests MUST fail when the Content-Type is
	224	+ // missing, because the query is defined by the request content
	225	+ // and its media type.
	226	+ if method == justapi_core::query_method()
	227	+ && !req.headers().contains_key(http::header::CONTENT_TYPE)
	228	+ {
	229	+ return Ok(json_response(
	230	+ StatusCode::BAD_REQUEST,
	231	+ r#"{"error":"QUERY requires a Content-Type header"}"#,
	232	+ ));
	233	+ }
	234	+
	235	+ let mut headers = Vec::new();
	236	+ for (name, value) in req.headers() {
	237	+ headers.push((name.as_str().as_bytes().to_vec(), value.as_bytes().to_vec()));
	238	+ }
	239	+
	240	+ let content_type = req
	241	+ .headers()
	242	+ .get(http::header::CONTENT_TYPE)
	243	+ .and_then(\|v\| v.to_str().ok())
	244	+ .map(\|s\| s.to_string());
	245	+ let is_multipart = content_type
	246	+ .as_deref()
	247	+ .unwrap_or("")
	248	+ .starts_with("multipart/form-data");
	249	+ let req_body = req.into_body();
	250	+ let (body_bytes, multipart_form_res) = if is_multipart {
	251	+ let ct = content_type.unwrap();
	252	+ let form = justapi_core::multipart::parse_multipart(req_body, &ct).await;
	253	+ (vec![], Some(form))
	254	+ } else {
	255	+ let b = http_body_util::Limited::new(req_body, 50 * 1024 * 1024)
	256	+ .collect()
	257	+ .await
	258	+ .map_err(\|e\| anyhow::anyhow!("Body too large or error: {}", e))?
	259	+ .to_bytes();
	260	+ (b.to_vec(), None)
	261	+ };
	262	+
	263	+ let matched = match router.at(&method, &path) {
	264	+ Ok(m) => m,
	265	+ Err(justapi_core::router::RouterError::MethodNotAllowed) => {
	266	+ return Ok(json_response(
	267	+ StatusCode::METHOD_NOT_ALLOWED,
	268	+ r#"{"error":"method not allowed"}"#,
	269	+ ))
	270	+ }
	271	+ Err(justapi_core::router::RouterError::NotFound) => {
	272	+ return Ok(json_response(
	273	+ StatusCode::NOT_FOUND,
	274	+ r#"{"error":"not found"}"#,
	275	+ ))
	276	+ }
	277	+ };
	278	+
	279	+ let handler_id = *matched.handler;
	280	+
	281	+ let path_params: Vec<(String, String)> = matched
	282	+ .params
	283	+ .iter()
	284	+ .map(\|(k, v)\| (k.to_string(), v.to_string()))
	285	+ .collect();
	286	+
	287	+ if let Some(Some(batcher)) = batchers.get(handler_id) {
	288	+ // Submit to batcher!
	289	+ let breq = BatchedReq {
	290	+ method: method.to_string(),
	291	+ path: path.clone(),
	292	+ path_params,
	293	+ query_string,
	294	+ headers,
	295	+ body: body_bytes.to_vec(),
	296	+ };
	297	+ let res = batcher
	298	+ .execute(breq)
	299	+ .await
	300	+ .map_err(\|e\| anyhow::anyhow!("{}", e))?;
	301	+ return Ok(nr_to_response(res));
	302	+ }
	303	+
	304	+ // Rust-side JSON Schema validation (fast path, no Python round-trip).
	305	+ if let Some(Some(schema_json)) = schema_jsons.get(handler_id) {
	306	+ if let Err(verr) =
	307	+ justapi_core::validate::validate_json_schema(&body_bytes, schema_json)
	308	+ {
	309	+ let error_body = serde_json::json!({
	310	+ "type": "https://justapi.dev/errors/validation",
	311	+ "title": "Validation Error",
	312	+ "status": 422,
	313	+ "detail": verr.to_string(),
	314	+ "errors": verr.errors.iter().map(\|e\| {
	315	+ serde_json::json!({
	316	+ "field": e.field,
	317	+ "message": e.message,
	318	+ })
	319	+ }).collect::<Vec<_>>(),
	320	+ })
	321	+ .to_string();
	322	+ let mut resp = Response::new(ResponseBody::new(
	323	+ http_body_util::Full::new(Bytes::from(error_body))
	324	+ .map_err(\|e: std::convert::Infallible\| -> anyhow::Error { match e {} }),
	325	+ ));
	326	+ *resp.status_mut() = StatusCode::UNPROCESSABLE_ENTITY;
	327	+ resp.headers_mut().insert(
	328	+ http::header::CONTENT_TYPE,
	329	+ http::HeaderValue::from_static("application/problem+json"),
	330	+ );
	331	+ return Ok(resp);
	332	+ }
	333	+ }
	334	+
	335	+ // Auto-transaction for write methods
	336	+ let is_write = matches!(method, Method::POST \| Method::PUT \| Method::DELETE);
	337	+ let tx = if is_write {
	338	+ if let Some(ref pool) = *db_pool {
	339	+ match pool.begin().await {
	340	+ Ok(tx) => Some(tx),
	341	+ Err(e) => {
	342	+ tracing::error!("Failed to begin transaction: {}", e);
	343	+ return Ok(json_response(
	344	+ StatusCode::INTERNAL_SERVER_ERROR,
	345	+ r#"{"error":"database error"}"#,
	346	+ ));
	347	+ }
	348	+ }
	349	+ } else {
	350	+ None
	351	+ }
	352	+ } else {
	353	+ None
	354	+ };
	355	+
	356	+ // Offload Python execution (which acquires the GIL and blocks) to the blocking pool
	357	+ // to avoid starving Tokio's I/O worker threads.
	358	+ let handlers_clone = handlers.clone();
	359	+ let schemas_clone = schemas.clone();
	360	+ let db_url_str = db_url.clone();
	361	+ let trace_ctx = justapi_core::trace_context::get_current_trace_context();
	362	+
	363	+ let nr = tokio::task::spawn_blocking(move \|\| {
	364	+ Python::attach(\|py\| {
	365	+ let mut form_dict_py: Option<pyo3::Py<pyo3::types::PyDict>> = None;
	366	+ if let Some(Ok(form)) = multipart_form_res {
	367	+ let d = pyo3::types::PyDict::new(py);
	368	+ for (k, v) in form.fields.iter() {
	369	+ d.set_item(k, v).unwrap();
	370	+ }
	371	+ for f in form.files.iter() {
	372	+ let upload_file = crate::multipart::UploadFile::new(
	373	+ f.filename.clone(),
	374	+ f.content_type.clone(),
	375	+ f.temp_path.clone(),
	376	+ );
	377	+ let p = pyo3::Bound::new(py, upload_file).unwrap();
	378	+ d.set_item(&f.field_name, p).unwrap();
	379	+ }
	380	+ form_dict_py = Some(d.into());
	381	+ }
	382	+
	383	+ call_python_handler(
	384	+ py,
	385	+ &handlers_clone[handler_id],
	386	+ schemas_clone[handler_id].as_ref(),
	387	+ &path_params,
	388	+ &query_string,
	389	+ &headers,
	390	+ &body_bytes,
	391	+ db_url_str.as_deref(),
	392	+ trace_ctx,
	393	+ form_dict_py,
	394	+ method.as_str(),
	395	+ &path,
	396	+ )
	397	+ })
	398	+ })
	399	+ .await
	400	+ .map_err(\|e\| anyhow::anyhow!("Spawn blocking error: {}", e))?;
	401	+
	402	+ // Commit or rollback transaction
	403	+ if let Some(tx) = tx {
	404	+ if nr.status >= 200 && nr.status < 300 {
	405	+ if let Err(e) = tx.commit().await {
	406	+ tracing::error!("Failed to commit transaction: {}", e);
	407	+ }
	408	+ } else {
	409	+ if let Err(e) = tx.rollback().await {
	410	+ tracing::error!("Failed to rollback transaction: {}", e);
	411	+ }
	412	+ }
	413	+ }
	414	+
	415	+ Ok(nr_to_response(nr))
	416	+ })
	417	+ })
	418	+}
	419	+
	420	+/// Build a handler for the test client.
	421	+pub fn make_test_handler(
	422	+ router: Arc<Router<usize>>,
	423	+ handlers: Arc<Vec<Py<PyAny>>>,
	424	+ schemas: Arc<Vec<Option<Py<PyAny>>>>,
	425	+ schema_jsons: Arc<Vec<Option<String>>>,
	426	+ db_pool: Option<AnyPool>,
	427	+ db_url_str: Option<String>,
	428	+) -> justapi_core::middleware::HandlerFn {
	429	+ let db_pool = Arc::new(db_pool);
	430	+ let db_url = Arc::new(db_url_str);
	431	+ Arc::new(move \|req: hyper::Request<hyper::body::Incoming>\| {
	432	+ let router = router.clone();
	433	+ let handlers = handlers.clone();
	434	+ let schemas = schemas.clone();
	435	+ let schema_jsons = schema_jsons.clone();
	436	+ let db_pool = db_pool.clone();
	437	+ let db_url = db_url.clone();
	438	+ Box::pin(async move {
	439	+ let method = req.method().clone();
	440	+ let path = req.uri().path().to_string();
	441	+ let query_string = req.uri().query().unwrap_or("").as_bytes().to_vec();
	442	+
	443	+ // RFC 10008: QUERY requests MUST fail when the Content-Type is
	444	+ // missing, because the query is defined by the request content
	445	+ // and its media type.
	446	+ if method == justapi_core::query_method()
	447	+ && !req.headers().contains_key(http::header::CONTENT_TYPE)
	448	+ {
	449	+ return Ok(json_response(
	450	+ StatusCode::BAD_REQUEST,
	451	+ r#"{"error":"QUERY requires a Content-Type header"}"#,
	452	+ ));
	453	+ }
	454	+
	455	+ let mut headers = Vec::new();
	456	+ for (name, value) in req.headers() {
	457	+ headers.push((name.as_str().as_bytes().to_vec(), value.as_bytes().to_vec()));
	458	+ }
	459	+
	460	+ let body_bytes = http_body_util::Limited::new(req.into_body(), 50 * 1024 * 1024)
	461	+ .collect()
	462	+ .await
	463	+ .map_err(\|e\| anyhow::anyhow!("Body too large or error: {}", e))?
	464	+ .to_bytes();
	465	+
	466	+ let matched = match router.at(&method, &path) {
	467	+ Ok(m) => m,
	468	+ Err(justapi_core::router::RouterError::MethodNotAllowed) => {
	469	+ return Ok(justapi_core::json_response(
	470	+ hyper::StatusCode::METHOD_NOT_ALLOWED,
	471	+ r#"{"error":"method not allowed"}"#,
	472	+ ))
	473	+ }
	474	+ Err(justapi_core::router::RouterError::NotFound) => {
	475	+ return Ok(justapi_core::json_response(
	476	+ hyper::StatusCode::NOT_FOUND,
	477	+ r#"{"error":"not found"}"#,
	478	+ ))
	479	+ }
	480	+ };
	481	+
	482	+ let handler_id = *matched.handler;
	483	+
	484	+ // Rust-side JSON Schema validation
	485	+ if let Some(Some(schema_json)) = schema_jsons.get(handler_id) {
	486	+ if let Err(verr) =
	487	+ justapi_core::validate::validate_json_schema(&body_bytes, schema_json)
	488	+ {
	489	+ let error_body = serde_json::json!({
	490	+ "type": "https://justapi.dev/errors/validation",
	491	+ "title": "Validation Error",
	492	+ "status": 422,
	493	+ "detail": verr.to_string(),
	494	+ "errors": verr.errors.iter().map(\|e\| {
	495	+ serde_json::json!({
	496	+ "field": e.field,
	497	+ "message": e.message,
	498	+ })
	499	+ }).collect::<Vec<_>>(),
	500	+ })
	501	+ .to_string();
	502	+ let mut resp = hyper::Response::new(justapi_core::ResponseBody::new(
	503	+ http_body_util::Full::new(hyper::body::Bytes::from(error_body))
	504	+ .map_err(\|e: std::convert::Infallible\| -> anyhow::Error { match e {} }),
	505	+ ));
	506	+ *resp.status_mut() = hyper::StatusCode::UNPROCESSABLE_ENTITY;
	507	+ resp.headers_mut().insert(
	508	+ http::header::CONTENT_TYPE,
	509	+ http::HeaderValue::from_static("application/problem+json"),
	510	+ );
	511	+ return Ok(resp);
	512	+ }
	513	+ }
	514	+
	515	+ let path_params: Vec<(String, String)> = matched
	516	+ .params
	517	+ .iter()
	518	+ .map(\|(k, v)\| (k.to_string(), v.to_string()))
	519	+ .collect();
	520	+
	521	+ // Auto-transaction for write methods
	522	+ let is_write = matches!(method, Method::POST \| Method::PUT \| Method::DELETE);
	523	+ let tx = if is_write {
	524	+ if let Some(ref pool) = *db_pool {
	525	+ match pool.begin().await {
	526	+ Ok(tx) => Some(tx),
	527	+ Err(e) => {
	528	+ tracing::error!("Failed to begin transaction: {}", e);
	529	+ return Ok(justapi_core::json_response(
	530	+ hyper::StatusCode::INTERNAL_SERVER_ERROR,
	531	+ r#"{"error":"database error"}"#,
	532	+ ));
	533	+ }
	534	+ }
	535	+ } else {
	536	+ None
	537	+ }
	538	+ } else {
	539	+ None
	540	+ };
	541	+
	542	+ // Offload Python execution (which acquires the GIL and blocks) to the blocking pool
	543	+ // to avoid starving Tokio's I/O worker threads.
	544	+ let handlers_clone = handlers.clone();
	545	+ let schemas_clone = schemas.clone();
	546	+ let db_url_str = db_url.clone();
	547	+ let trace_ctx = justapi_core::trace_context::get_current_trace_context();
	548	+
	549	+ let method_clone = method.to_string();
	550	+ let path_clone = path.clone();
	551	+ let nr = tokio::task::spawn_blocking(move \|\| {
	552	+ Python::attach(\|py\| {
	553	+ call_python_handler(
	554	+ py,
	555	+ &handlers_clone[handler_id],
	556	+ schemas_clone[handler_id].as_ref(),
	557	+ &path_params,
	558	+ &query_string,
	559	+ &headers,
	560	+ &body_bytes,
	561	+ db_url_str.as_deref(),
	562	+ trace_ctx,
	563	+ None,
	564	+ method_clone.as_str(),
	565	+ &path_clone,
	566	+ )
	567	+ })
	568	+ })
	569	+ .await
	570	+ .map_err(\|e\| anyhow::anyhow!("Spawn blocking error: {}", e))?;
	571	+
	572	+ // Commit or rollback transaction
	573	+ if let Some(tx) = tx {
	574	+ if nr.status >= 200 && nr.status < 300 {
	575	+ if let Err(e) = tx.commit().await {
	576	+ tracing::error!("Failed to commit transaction: {}", e);
	577	+ }
	578	+ } else {
	579	+ if let Err(e) = tx.rollback().await {
	580	+ tracing::error!("Failed to rollback transaction: {}", e);
	581	+ }
	582	+ }
	583	+ }
	584	+
	585	+ Ok(nr_to_response(nr))
	586	+ })
	587	+ })
	588	+}
	589	+
	590	+pub(crate) fn call_plugin_hook(py: Python<'_>, plugin: &Py<PyAny>, hook_name: &str) -> PyResult<()> {
	591	+ let helper = get_helper(py);
	592	+ helper
	593	+ .call_plugin_hook
	594	+ .bind(py)
	595	+ .call1((plugin, hook_name))?;
	596	+ Ok(())
	597	+}
	598	+
	599	+/// Resolve a Python schema object to a JSON Schema string.
	600	+///
	601	+/// Accepts:
	602	+/// - A string (already a JSON Schema)
	603	+/// - A Schema subclass (has `_schema_json()` method)
	604	+/// - A Pydantic model (has `model_json_schema()` method)
	605	+pub(crate) fn resolve_schema_json(py: Python<'_>, schema: Option<Py<PyAny>>) -> PyResult<Option<String>> {
	606	+ let Some(schema) = schema else {
	607	+ return Ok(None);
	608	+ };
	609	+
	610	+ let schema_bound = schema.bind(py);
	611	+
	612	+ // Check if it's a string
	613	+ if let Ok(s) = schema_bound.extract::<String>() {
	614	+ return Ok(Some(s));
	615	+ }
	616	+
	617	+ // Try calling _schema_json() (our Schema class)
	618	+ if let Ok(method) = schema_bound.getattr("_schema_json") {
	619	+ if let Ok(s) = method.call0()?.extract::<String>() {
	620	+ return Ok(Some(s));
	621	+ }
	622	+ }
	623	+
	624	+ // Try Pydantic v2's model_json_schema()
	625	+ if let Ok(method) = schema_bound.getattr("model_json_schema") {
	626	+ if let Ok(schema_dict) = method.call0() {
	627	+ let json_mod = PyModule::import(py, "json")?;
	628	+ let s = json_mod
	629	+ .getattr("dumps")?
	630	+ .call1((schema_dict,))?
	631	+ .extract::<String>()?;
	632	+ return Ok(Some(s));
	633	+ }
	634	+ }
	635	+
	636	+ // Try Pydantic v1's schema()
	637	+ if let Ok(method) = schema_bound.getattr("schema") {
	638	+ if let Ok(schema_dict) = method.call0() {
	639	+ let json_mod = PyModule::import(py, "json")?;
	640	+ let s = json_mod
	641	+ .getattr("dumps")?
	642	+ .call1((schema_dict,))?
	643	+ .extract::<String>()?;
	644	+ return Ok(Some(s));
	645	+ }
	646	+ }
	647	+
	648	+ Err(pyo3::exceptions::PyTypeError::new_err(
	649	+ "schema must be a JSON Schema string, Schema subclass, or Pydantic model",
	650	+ ))
	651	+}
0	652

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

+9-0

@@ -0,0 +1,9 @@
	1	+pub mod types;
	2	+pub mod handlers;
	3	+pub mod app;
	4	+
	5	+pub use types::*;
	6	+pub use handlers::*;
	7	+pub use app::*;
	8	+
	9	+pub const NATIVE_HELPER: &str = include_str!("../../python/justapi/_native_helper.py");
0	10

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

+138-0

@@ -0,0 +1,138 @@
	1	+use std::sync::OnceLock;
	2	+
	3	+use hyper::body::Bytes;
	4	+use pyo3::prelude::*;
	5	+
	6	+use super::NATIVE_HELPER;
	7	+/// Response from a native Python handler.
	8	+pub enum NativeBody {
	9	+ Bytes(Vec<u8>),
	10	+ Stream(tokio::sync::mpsc::UnboundedReceiver<Result<Bytes, anyhow::Error>>),
	11	+}
	12	+
	13	+pub struct NativeResponse {
	14	+ pub status: u16,
	15	+ pub headers: Vec<(Vec<u8>, Vec<u8>)>,
	16	+ pub body: NativeBody,
	17	+}
	18	+
	19	+#[pyclass(name = "_StreamSender")]
	20	+pub struct StreamSender {
	21	+ pub(crate) tx: tokio::sync::mpsc::UnboundedSender<Result<Bytes, anyhow::Error>>,
	22	+}
	23	+
	24	+#[pymethods]
	25	+impl StreamSender {
	26	+ fn send(&self, data: &[u8]) {
	27	+ let _ = self.tx.send(Ok(Bytes::copy_from_slice(data)));
	28	+ }
	29	+
	30	+ fn send_error(&self, msg: String) {
	31	+ let _ = self.tx.send(Err(anyhow::anyhow!("{}", msg)));
	32	+ }
	33	+
	34	+ fn close(&self) {}
	35	+}
	36	+
	37	+#[pyclass(name = "TokenStreamResponse", subclass)]
	38	+pub struct TokenStreamResponse {
	39	+ pub generator: Py<PyAny>,
	40	+ pub status: u16,
	41	+ pub headers: Vec<(Vec<u8>, Vec<u8>)>,
	42	+}
	43	+
	44	+#[pymethods]
	45	+impl TokenStreamResponse {
	46	+ #[new]
	47	+ #[pyo3(signature = (generator, status=200, headers=None))]
	48	+ pub fn new(
	49	+ generator: Py<PyAny>,
	50	+ status: u16,
	51	+ headers: Option<Vec<(Vec<u8>, Vec<u8>)>>,
	52	+ ) -> Self {
	53	+ let headers = headers
	54	+ .unwrap_or_else(\|\| vec![(b"content-type".to_vec(), b"text/event-stream".to_vec())]);
	55	+ Self {
	56	+ generator,
	57	+ status,
	58	+ headers,
	59	+ }
	60	+ }
	61	+}
	62	+
	63	+#[derive(Clone)]
	64	+pub struct BatchedReq {
	65	+ pub path: String,
	66	+ pub method: String,
	67	+ pub path_params: Vec<(String, String)>,
	68	+ pub query_string: Vec<u8>,
	69	+ pub headers: Vec<(Vec<u8>, Vec<u8>)>,
	70	+ pub body: Vec<u8>,
	71	+}
	72	+
	73	+pub(crate) struct HelperFunctions {
	74	+ pub(crate) call_handler: Py<PyAny>,
	75	+ pub(crate) call_batch_handler: Py<PyAny>,
	76	+ pub(crate) validate_body: Py<PyAny>,
	77	+ pub(crate) call_plugin_hook: Py<PyAny>,
	78	+ pub(crate) wrap_result: Py<PyAny>,
	79	+ pub(crate) pump_stream: Py<PyAny>,
	80	+ pub(crate) run_ws_handler: Py<PyAny>,
	81	+ pub(crate) set_trace_context: Py<PyAny>,
	82	+}
	83	+
	84	+pub(crate) fn get_helper(py: Python<'_>) -> &HelperFunctions {
	85	+ static HELPER: OnceLock<HelperFunctions> = OnceLock::new();
	86	+ HELPER.get_or_init(\|\| {
	87	+ let code = std::ffi::CString::new(NATIVE_HELPER).expect("valid CString");
	88	+ let filename = std::ffi::CString::new("_native_helper.py").expect("valid CString");
	89	+ let name = std::ffi::CString::new("_native_helper").expect("valid CString");
	90	+
	91	+ let helper = PyModule::from_code(py, code.as_c_str(), filename.as_c_str(), name.as_c_str())
	92	+ .expect("Native helper module should compile");
	93	+
	94	+ let call_handler = helper
	95	+ .getattr("call_handler")
	96	+ .expect("call_handler function should exist")
	97	+ .unbind();
	98	+ let call_batch_handler = helper
	99	+ .getattr("call_batch_handler")
	100	+ .expect("call_batch_handler function should exist")
	101	+ .unbind();
	102	+ let validate_body = helper
	103	+ .getattr("validate_body")
	104	+ .expect("validate_body function should exist")
	105	+ .unbind();
	106	+ let call_plugin_hook = helper
	107	+ .getattr("call_plugin_hook")
	108	+ .expect("call_plugin_hook function should exist")
	109	+ .unbind();
	110	+ let wrap_result = helper
	111	+ .getattr("wrap_result")
	112	+ .expect("wrap_result function should exist")
	113	+ .unbind();
	114	+ let pump_stream = helper
	115	+ .getattr("_pump_stream")
	116	+ .expect("_pump_stream function should exist")
	117	+ .unbind();
	118	+ let run_ws_handler = helper
	119	+ .getattr("run_ws_handler")
	120	+ .expect("run_ws_handler function should exist")
	121	+ .unbind();
	122	+ let set_trace_context = helper
	123	+ .getattr("set_trace_context")
	124	+ .expect("set_trace_context function should exist")
	125	+ .unbind();
	126	+
	127	+ HelperFunctions {
	128	+ call_handler,
	129	+ call_batch_handler,
	130	+ validate_body,
	131	+ call_plugin_hook,
	132	+ wrap_result,
	133	+ pump_stream,
	134	+ run_ws_handler,
	135	+ set_trace_context,
	136	+ }
	137	+ })
	138	+}
0	139

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

+59-0

@@ -0,0 +1,59 @@
	1	+use justapi_core::rate_limit::RateLimiter;
	2	+use pyo3::prelude::*;
	3	+use std::sync::Arc;
	4	+
	5	+#[pyclass(name = "RateLimiter")]
	6	+pub struct PyRateLimiter {
	7	+ inner: Arc<RateLimiter>,
	8	+}
	9	+
	10	+#[pymethods]
	11	+impl PyRateLimiter {
	12	+ #[staticmethod]
	13	+ pub fn new_redis<'py>(py: Python<'py>, redis_url: String) -> PyResult<Bound<'py, PyAny>> {
	14	+ let fut = async move {
	15	+ let limiter = RateLimiter::new_redis(&redis_url)
	16	+ .await
	17	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
	18	+ Ok(PyRateLimiter {
	19	+ inner: Arc::new(limiter),
	20	+ })
	21	+ };
	22	+ pyo3_async_runtimes::tokio::future_into_py(py, fut)
	23	+ }
	24	+
	25	+ pub fn check_limit<'py>(
	26	+ &self,
	27	+ py: Python<'py>,
	28	+ key: String,
	29	+ capacity: u64,
	30	+ replenish_rate: u64,
	31	+ tokens: u64,
	32	+ ) -> PyResult<Bound<'py, PyAny>> {
	33	+ let limiter = self.inner.clone();
	34	+ let fut = async move {
	35	+ let res = limiter
	36	+ .check_limit(&key, capacity, replenish_rate, tokens)
	37	+ .await
	38	+ .map_err(\|e\| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
	39	+ Ok(PyRateLimitResult {
	40	+ allowed: res.allowed,
	41	+ retry_after_ms: res.retry_after_ms,
	42	+ })
	43	+ };
	44	+ pyo3_async_runtimes::tokio::future_into_py(py, fut)
	45	+ }
	46	+}
	47	+
	48	+#[pyclass(name = "RateLimitResult", get_all, from_py_object)]
	49	+#[derive(Clone)]
	50	+pub struct PyRateLimitResult {
	51	+ pub allowed: bool,
	52	+ pub retry_after_ms: u64,
	53	+}
	54	+
	55	+pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
	56	+ m.add_class::<PyRateLimiter>()?;
	57	+ m.add_class::<PyRateLimitResult>()?;
	58	+ Ok(())
	59	+}
0	60

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

+166-0

@@ -0,0 +1,166 @@
	1	+use pyo3::prelude::*;
	2	+use pyo3::types::{PyAny, PyBytes, PyDict, PyList};
	3	+
	4	+#[pyclass(mapping)]
	5	+pub struct Request {
	6	+ method: String,
	7	+ path: String,
	8	+ path_params_raw: Vec<(String, String)>,
	9	+ query_string_raw: Vec<u8>,
	10	+ headers_raw: Vec<(Vec<u8>, Vec<u8>)>,
	11	+ body_raw: Vec<u8>,
	12	+ db_url_raw: Option<String>,
	13	+
	14	+ state: Py<PyDict>,
	15	+ path_params_cached: Option<Py<PyDict>>,
	16	+ form_data: Option<Py<PyDict>>,
	17	+}
	18	+
	19	+impl Request {
	20	+ #[allow(clippy::too_many_arguments)]
	21	+ pub fn new(
	22	+ py: Python<'_>,
	23	+ method: String,
	24	+ path: String,
	25	+ path_params: Vec<(String, String)>,
	26	+ query_string: Vec<u8>,
	27	+ headers: Vec<(Vec<u8>, Vec<u8>)>,
	28	+ body: Vec<u8>,
	29	+ db_url: Option<String>,
	30	+ form_data: Option<Py<PyDict>>,
	31	+ ) -> Self {
	32	+ Self {
	33	+ method,
	34	+ path,
	35	+ path_params_raw: path_params,
	36	+ query_string_raw: query_string,
	37	+ headers_raw: headers,
	38	+ body_raw: body,
	39	+ db_url_raw: db_url,
	40	+ state: PyDict::new(py).into(),
	41	+ path_params_cached: None,
	42	+ form_data,
	43	+ }
	44	+ }
	45	+}
	46	+
	47	+#[pymethods]
	48	+impl Request {
	49	+ #[getter]
	50	+ fn method(&self) -> PyResult<String> {
	51	+ Ok(self.method.clone())
	52	+ }
	53	+
	54	+ #[getter]
	55	+ fn path(&self) -> PyResult<String> {
	56	+ Ok(self.path.clone())
	57	+ }
	58	+
	59	+ #[getter]
	60	+ fn path_params(&mut self, py: Python<'_>) -> PyResult<Py<PyDict>> {
	61	+ if let Some(ref d) = self.path_params_cached {
	62	+ return Ok(d.clone_ref(py));
	63	+ }
	64	+ let d = PyDict::new(py);
	65	+ for (k, v) in &self.path_params_raw {
	66	+ d.set_item(k.as_str(), v.as_str())?;
	67	+ }
	68	+ let ret: Py<PyDict> = d.into();
	69	+ self.path_params_cached = Some(ret.clone_ref(py));
	70	+ Ok(ret)
	71	+ }
	72	+
	73	+ #[getter]
	74	+ fn headers(&self, py: Python<'_>) -> PyResult<Py<PyList>> {
	75	+ let l = PyList::empty(py);
	76	+ for (k, v) in &self.headers_raw {
	77	+ l.append((PyBytes::new(py, k), PyBytes::new(py, v)))?;
	78	+ }
	79	+ Ok(l.into())
	80	+ }
	81	+
	82	+ #[getter]
	83	+ fn query_string(&self, py: Python<'_>) -> PyResult<Py<PyBytes>> {
	84	+ Ok(PyBytes::new(py, &self.query_string_raw).into())
	85	+ }
	86	+
	87	+ #[getter]
	88	+ fn body(&self, py: Python<'_>) -> PyResult<Py<PyBytes>> {
	89	+ Ok(PyBytes::new(py, &self.body_raw).into())
	90	+ }
	91	+
	92	+ fn json(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
	93	+ let json_module = py.import("json")?;
	94	+ let body_bytes = PyBytes::new(py, &self.body_raw);
	95	+ let parsed = json_module.getattr("loads")?.call1((body_bytes,))?;
	96	+ Ok(parsed.into())
	97	+ }
	98	+
	99	+ #[pyo3(signature = (key, default=None))]
	100	+ fn get(
	101	+ &mut self,
	102	+ py: Python<'_>,
	103	+ key: &str,
	104	+ default: Option<Py<PyAny>>,
	105	+ ) -> PyResult<Py<PyAny>> {
	106	+ match key {
	107	+ "method" => Ok(pyo3::types::PyString::new(py, &self.method)
	108	+ .into_any()
	109	+ .into()),
	110	+ "path" => Ok(pyo3::types::PyString::new(py, &self.path).into_any().into()),
	111	+ "path_params" => Ok(self.path_params(py)?.into_any()),
	112	+ "query_string" => Ok(self.query_string(py)?.into_any()),
	113	+ "headers" => Ok(self.headers(py)?.into_any()),
	114	+ "body" => Ok(self.body(py)?.into_any()),
	115	+ "db_url" => {
	116	+ if let Some(db) = &self.db_url_raw {
	117	+ let py_str = pyo3::types::PyString::new(py, db);
	118	+ Ok(py_str.into_any().into())
	119	+ } else {
	120	+ Ok(py.None())
	121	+ }
	122	+ }
	123	+ "form" => {
	124	+ if let Some(ref d) = self.form_data {
	125	+ Ok(d.clone_ref(py).into_any())
	126	+ } else {
	127	+ Ok(py.None())
	128	+ }
	129	+ }
	130	+ _ => {
	131	+ let state = self.state.bind(py);
	132	+ if let Some(v) = state.get_item(key)? {
	133	+ Ok(v.into())
	134	+ } else {
	135	+ Ok(default.unwrap_or_else(\|\| py.None()))
	136	+ }
	137	+ }
	138	+ }
	139	+ }
	140	+
	141	+ fn __getitem__(&mut self, py: Python<'_>, key: &str) -> PyResult<Py<PyAny>> {
	142	+ let val = self.get(py, key, None)?;
	143	+ if val.is_none(py) {
	144	+ Err(pyo3::exceptions::PyKeyError::new_err(key.to_string()))
	145	+ } else {
	146	+ Ok(val)
	147	+ }
	148	+ }
	149	+
	150	+ fn __setitem__(&mut self, py: Python<'_>, key: &str, value: Py<PyAny>) -> PyResult<()> {
	151	+ let state = self.state.bind(py);
	152	+ state.set_item(key, value)?;
	153	+ Ok(())
	154	+ }
	155	+
	156	+ fn __contains__(&mut self, py: Python<'_>, key: &str) -> PyResult<bool> {
	157	+ let state = self.state.bind(py);
	158	+ if state.contains(key)? {
	159	+ return Ok(true);
	160	+ }
	161	+ match key {
	162	+ "path_params" \| "query_string" \| "headers" \| "body" \| "db_url" => Ok(true),
	163	+ _ => Ok(false),
	164	+ }
	165	+ }
	166	+}
0	167

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

+1-0

@@ -0,0 +1 @@
	1	+fn main() {}
0	2

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

+177-0

@@ -0,0 +1,177 @@
	1	+//! Python JustAPITestClient — wraps Rust TestClient via PyO3.
	2	+//!
	3	+//! Usage:
	4	+//! ```python
	5	+//! from justapi import JustAPIApp, JustAPITestClient
	6	+//!
	7	+//! app = JustAPIApp()
	8	+//! app.get("/hello", lambda r: {"message": "hello"})
	9	+//!
	10	+//! client = JustAPITestClient(app)
	11	+//! resp = client.get("/hello")
	12	+//! assert resp.status == 200
	13	+//!
	14	+//! # With database:
	15	+//! client = JustAPITestClient(app, database="sqlite::memory:")
	16	+//! resp = client.post("/users", b'{"name":"Alice"}')
	17	+//! assert resp.status == 201
	18	+//! ```
	19	+
	20	+use std::sync::Arc;
	21	+
	22	+use pyo3::prelude::*;
	23	+use pyo3::types::*;
	24	+
	25	+use hyper::header::HeaderName;
	26	+use justapi_core::coalesce::RequestCoalescer;
	27	+use justapi_core::db::{AnyPool, DatabaseConfig, PoolManager};
	28	+use justapi_core::middleware::{HandlerFn, MiddlewareChain};
	29	+use justapi_core::testing::TestClient;
	30	+
	31	+use crate::native::JustAPIApp;
	32	+
	33	+/// Python test client for JustAPI.
	34	+///
	35	+/// Sends requests through the full JustAPI pipeline
	36	+/// without starting a TCP server.
	37	+#[pyclass]
	38	+pub struct JustAPITestClient {
	39	+ test_client: Option<TestClient>,
	40	+ #[allow(dead_code)]
	41	+ db_pool: Option<AnyPool>,
	42	+}
	43	+
	44	+#[pymethods]
	45	+impl JustAPITestClient {
	46	+ #[new]
	47	+ #[pyo3(signature = (app, database=None))]
	48	+ fn new(app: &mut JustAPIApp, py: Python<'_>, database: Option<String>) -> PyResult<Self> {
	49	+ let router = Arc::new(std::mem::take(&mut app.router));
	50	+ let handlers = Arc::new(std::mem::take(&mut app.handlers));
	51	+ let schemas = Arc::new(std::mem::take(&mut app.schemas));
	52	+ let schema_jsons = Arc::new(std::mem::take(&mut app.schema_jsons));
	53	+
	54	+ // Initialize database pool if a database URL is provided.
	55	+ let (db_pool, db_url_str) = if let Some(ref url) = database {
	56	+ let config = DatabaseConfig {
	57	+ kind: Some(justapi_core::db::DbKind::from_url(url)),
	58	+ url: url.clone(),
	59	+ max_connections: 5,
	60	+ };
	61	+ let pool = py
	62	+ .detach(move \|\| -> Result<AnyPool, String> {
	63	+ let rt = tokio::runtime::Runtime::new().map_err(\|e\| e.to_string())?;
	64	+ rt.block_on(async {
	65	+ let mut mgr = PoolManager::new();
	66	+ mgr.init("test", config)
	67	+ .await
	68	+ .map_err(\|e\| format!("Database init error: {}", e))
	69	+ })
	70	+ })
	71	+ .map_err(\|e: String\| pyo3::exceptions::PyRuntimeError::new_err(e))?;
	72	+ (Some(pool), database)
	73	+ } else {
	74	+ (None, None)
	75	+ };
	76	+
	77	+ let handler = super::native::make_test_handler(
	78	+ router,
	79	+ handlers,
	80	+ schemas,
	81	+ schema_jsons,
	82	+ db_pool.clone(),
	83	+ db_url_str,
	84	+ );
	85	+
	86	+ // Apply request coalescing (singleflight) if the app enabled it, so the
	87	+ // test client exercises the same pipeline as the real server.
	88	+ let coalesce_headers = app.coalesce_headers.take();
	89	+ let handler: HandlerFn = match coalesce_headers {
	90	+ Some(headers) => {
	91	+ let mut coalescer = RequestCoalescer::new();
	92	+ if !headers.is_empty() {
	93	+ let mut names = Vec::new();
	94	+ for h in &headers {
	95	+ if let Ok(n) = HeaderName::from_bytes(h.as_bytes()) {
	96	+ names.push(n);
	97	+ }
	98	+ }
	99	+ coalescer = coalescer.with_headers(&names);
	100	+ }
	101	+ let mut chain = MiddlewareChain::new(handler);
	102	+ chain.add(coalescer);
	103	+ Arc::new(move \|req\| {
	104	+ let c = chain.clone();
	105	+ Box::pin(async move { c.run(req).await })
	106	+ })
	107	+ }
	108	+ None => handler,
	109	+ };
	110	+
	111	+ let test_client = TestClient::new(handler);
	112	+ Ok(Self {
	113	+ test_client: Some(test_client),
	114	+ db_pool,
	115	+ })
	116	+ }
	117	+
	118	+ fn get(&self, py: Python<'_>, path: &str) -> PyResult<Py<PyAny>> {
	119	+ let path = path.to_owned();
	120	+ self.do_request(py, move \|client, rt\| rt.block_on(client.get(&path)))
	121	+ }
	122	+
	123	+ fn post(&self, py: Python<'_>, path: &str, body: Vec<u8>) -> PyResult<Py<PyAny>> {
	124	+ let path = path.to_owned();
	125	+ self.do_request(py, move \|client, rt\| rt.block_on(client.post(&path, body)))
	126	+ }
	127	+
	128	+ fn put(&self, py: Python<'_>, path: &str, body: Vec<u8>) -> PyResult<Py<PyAny>> {
	129	+ let path = path.to_owned();
	130	+ self.do_request(py, move \|client, rt\| rt.block_on(client.put(&path, body)))
	131	+ }
	132	+
	133	+ fn delete(&self, py: Python<'_>, path: &str) -> PyResult<Py<PyAny>> {
	134	+ let path = path.to_owned();
	135	+ self.do_request(py, move \|client, rt\| rt.block_on(client.delete(&path)))
	136	+ }
	137	+}
	138	+
	139	+impl JustAPITestClient {
	140	+ fn do_request<F>(&self, py: Python<'_>, f: F) -> PyResult<Py<PyAny>>
	141	+ where
	142	+ F: FnOnce(
	143	+ TestClient,
	144	+ &tokio::runtime::Runtime,
	145	+ ) -> Result<justapi_core::testing::TestResponse, anyhow::Error>
	146	+ + Send,
	147	+ {
	148	+ let client = self.test_client.clone().ok_or_else(\|\| {
	149	+ pyo3::exceptions::PyRuntimeError::new_err("TestClient already consumed")
	150	+ })?;
	151	+
	152	+ let resp = py
	153	+ .detach(move \|\| {
	154	+ let rt =
	155	+ tokio::runtime::Runtime::new().map_err(\|e\| format!("Runtime error: {}", e))?;
	156	+ f(client, &rt).map_err(\|e\| format!("Request failed: {}", e))
	157	+ })
	158	+ .map_err(\|e: String\| pyo3::exceptions::PyRuntimeError::new_err(e))?;
	159	+
	160	+ Ok(response_to_py(&resp))
	161	+ }
	162	+}
	163	+
	164	+fn response_to_py(resp: &justapi_core::testing::TestResponse) -> Py<PyAny> {
	165	+ Python::<'_>::try_attach(\|py\| {
	166	+ let d = PyDict::new(py);
	167	+ d.set_item("status", resp.status).ok();
	168	+ d.set_item("body", &resp.body[..]).ok();
	169	+ let hdrs = PyDict::new(py);
	170	+ for (k, v) in &resp.headers {
	171	+ hdrs.set_item(k.as_str(), v.as_str()).ok();
	172	+ }
	173	+ d.set_item("headers", hdrs).ok();
	174	+ d.into_any().unbind()
	175	+ })
	176	+ .expect("GIL already held in test client context")
	177	+}
0	178

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

+141-0

@@ -0,0 +1,141 @@
	1	+use std::sync::Arc;
	2	+
	3	+use pyo3::exceptions::{PyConnectionError, PyEOFError};
	4	+use pyo3::prelude::*;
	5	+use tokio::sync::mpsc::UnboundedSender;
	6	+use tokio::sync::Mutex;
	7	+
	8	+/// Internal message exchanged between the Rust bridging task and the Python
	9	+/// `WebSocket` object.
	10	+pub enum WsMessage {
	11	+ Text(String),
	12	+ Bytes(Vec<u8>),
	13	+ Close,
	14	+}
	15	+
	16	+/// A WebSocket connection exposed to Python handlers.
	17	+///
	18	+/// Instances are created by JustAPIApp when a WebSocket upgrade is accepted and
	19	+/// passed as the single argument to a `@app.websocket()` handler:
	20	+///
	21	+/// ```python
	22	+/// @app.websocket("/ws")
	23	+/// async def handler(ws):
	24	+/// await ws.accept()
	25	+/// await ws.send_text("hello")
	26	+/// msg = await ws.receive_text()
	27	+/// await ws.close()
	28	+/// ```
	29	+///
	30	+/// All methods are async and integrate with the daemon asyncio event loop via
	31	+/// `pyo3-async-runtimes`, so they can be `await`ed directly inside an async
	32	+/// handler. The actual bytes cross the PyO3 boundary through lock-free channels
	33	+/// that the Rust side bridges to the tokio-tungstenite stream.
	34	+#[pyclass]
	35	+pub struct WebSocket {
	36	+ incoming: Arc<Mutex<tokio::sync::mpsc::UnboundedReceiver<WsMessage>>>,
	37	+ outgoing: UnboundedSender<WsMessage>,
	38	+}
	39	+
	40	+impl WebSocket {
	41	+ pub fn new(
	42	+ incoming: Arc<Mutex<tokio::sync::mpsc::UnboundedReceiver<WsMessage>>>,
	43	+ outgoing: UnboundedSender<WsMessage>,
	44	+ ) -> Self {
	45	+ Self { incoming, outgoing }
	46	+ }
	47	+}
	48	+
	49	+#[pymethods]
	50	+impl WebSocket {
	51	+ /// Complete the handshake. The Rust side has already accepted the upgrade,
	52	+ /// so this is effectively a no-op kept for API parity with other frameworks.
	53	+ fn accept<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
	54	+ pyo3_async_runtimes::tokio::future_into_py(py, async { Ok(()) })
	55	+ }
	56	+
	57	+ /// Receive the next message as a `str`/`bytes`, or `None` when the peer
	58	+ /// closes the connection.
	59	+ fn receive<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
	60	+ let incoming = self.incoming.clone();
	61	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	62	+ let mut rx = incoming.lock().await;
	63	+ match rx.recv().await {
	64	+ Some(WsMessage::Text(t)) => Ok(Some(t)),
	65	+ Some(WsMessage::Bytes(b)) => Ok(Some(String::from_utf8_lossy(&b).into_owned())),
	66	+ Some(WsMessage::Close) \| None => Ok(None),
	67	+ }
	68	+ })
	69	+ }
	70	+
	71	+ /// Receive the next message as text, raising `EOFError` on close.
	72	+ fn receive_text<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
	73	+ let incoming = self.incoming.clone();
	74	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	75	+ let mut rx = incoming.lock().await;
	76	+ match rx.recv().await {
	77	+ Some(WsMessage::Text(t)) => Ok(t),
	78	+ Some(WsMessage::Bytes(b)) => Ok(String::from_utf8_lossy(&b).into_owned()),
	79	+ Some(WsMessage::Close) \| None => {
	80	+ Err(PyEOFError::new_err("websocket connection closed"))
	81	+ }
	82	+ }
	83	+ })
	84	+ }
	85	+
	86	+ /// Receive the next message as raw bytes, raising `EOFError` on close.
	87	+ fn receive_bytes<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
	88	+ let incoming = self.incoming.clone();
	89	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	90	+ let mut rx = incoming.lock().await;
	91	+ match rx.recv().await {
	92	+ Some(WsMessage::Text(t)) => Ok(t.into_bytes()),
	93	+ Some(WsMessage::Bytes(b)) => Ok(b),
	94	+ Some(WsMessage::Close) \| None => {
	95	+ Err(PyEOFError::new_err("websocket connection closed"))
	96	+ }
	97	+ }
	98	+ })
	99	+ }
	100	+
	101	+ /// Send a message. `bytes` are sent as binary.
	102	+ fn send<'py>(&self, py: Python<'py>, message: &[u8]) -> PyResult<Bound<'py, PyAny>> {
	103	+ self.send_bytes(py, message)
	104	+ }
	105	+
	106	+ /// Send a UTF-8 text message.
	107	+ fn send_text<'py>(&self, py: Python<'py>, message: String) -> PyResult<Bound<'py, PyAny>> {
	108	+ let outgoing = self.outgoing.clone();
	109	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	110	+ outgoing
	111	+ .send(WsMessage::Text(message))
	112	+ .map_err(\|_\| PyConnectionError::new_err("websocket connection closed"))?;
	113	+ Ok(())
	114	+ })
	115	+ }
	116	+
	117	+ /// Send a binary message.
	118	+ fn send_bytes<'py>(&self, py: Python<'py>, message: &[u8]) -> PyResult<Bound<'py, PyAny>> {
	119	+ let outgoing = self.outgoing.clone();
	120	+ let msg = message.to_vec();
	121	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	122	+ outgoing
	123	+ .send(WsMessage::Bytes(msg))
	124	+ .map_err(\|_\| PyConnectionError::new_err("websocket connection closed"))?;
	125	+ Ok(())
	126	+ })
	127	+ }
	128	+
	129	+ /// Send a close frame and terminate the connection.
	130	+ fn close<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
	131	+ let outgoing = self.outgoing.clone();
	132	+ pyo3_async_runtimes::tokio::future_into_py(py, async move {
	133	+ let _ = outgoing.send(WsMessage::Close);
	134	+ Ok(())
	135	+ })
	136	+ }
	137	+
	138	+ fn __repr__(&self) -> String {
	139	+ "WebSocket()".to_string()
	140	+ }
	141	+}
0	142