swadhin / justapi public

📦 p3: app.sse_native registration + sse_specs alignment (ADR-088)

swadhinbiswas committed 2 months ago
576a7be
1 parent d8c16dd
commit 576a7be
1 file changed
+97 -0
Showing 1 changed file+97-0
crates/justapi-py/src/native/app.rs
▼crates/justapi-py/src/native/app.rs
+97-0
@@ -53,6 +53,10 @@ pub struct JustAPIApp {
53
53
/// `crud_dispatch_bytes`, with no GIL hop. The operation is inferred from
54
54
/// the request method at runtime. Indexed by handler id.
55
55
pub crud: Vec<Option<(String, Vec<String>, String)>>,
56
+ /// Rust-native SSE stream spec per route: `Some((count, interval_ms))`
57
+ /// means the route serves a Rust-generated SSE stream with no Python
58
+ /// involvement (ADR-088). Indexed by handler id.
59
+ pub sse_specs: Vec<Option<(u64, u64)>>,
56
60
pub schemas: Vec<Option<Py<PyAny>>>,
57
61
pub schema_jsons: Vec<Option<String>>,
58
62
/// JSON Schemas (resolved at registration) used by the native fast path to
@@ -466,6 +470,7 @@ impl JustAPIApp {
466
470
handlers: Vec::new(),
467
471
native: Vec::new(),
468
472
crud: Vec::new(),
473
+ sse_specs: Vec::new(),
469
474
schemas: Vec::new(),
470
475
schema_jsons: Vec::new(),
471
476
query_schema_jsons: Vec::new(),
@@ -942,6 +947,7 @@ impl JustAPIApp {
942
947
self.handlers.push(handler);
943
948
self.native.push(is_native);
944
949
self.crud.push(crud_spec);
950
+ self.sse_specs.push(None);
945
951
self.schemas.push(None);
946
952
self.schema_jsons.push(None);
947
953
self.query_schema_jsons.push(query_json);
@@ -1004,6 +1010,7 @@ impl JustAPIApp {
1004
1010
self.handlers.push(handler);
1005
1011
self.native.push(native.unwrap_or(false));
1006
1012
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1013
+ self.sse_specs.push(None);
1007
1014
self.schemas.push(body_schema.as_ref().map(|b| b.clone_ref(py)));
1008
1015
self.schema_jsons.push(resolve_schema_json(py, schema)?);
1009
1016
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1063,6 +1070,7 @@ impl JustAPIApp {
1063
1070
self.handlers.push(handler);
1064
1071
self.native.push(native.unwrap_or(false));
1065
1072
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1073
+ self.sse_specs.push(None);
1066
1074
self.schemas.push(body_schema.as_ref().map(|b| b.clone_ref(py)));
1067
1075
self.schema_jsons.push(resolve_schema_json(py, schema)?);
1068
1076
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1122,6 +1130,7 @@ impl JustAPIApp {
1122
1130
self.handlers.push(handler);
1123
1131
self.native.push(native.unwrap_or(false));
1124
1132
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1133
+ self.sse_specs.push(None);
1125
1134
self.schemas.push(body_schema.as_ref().map(|b| b.clone_ref(py)));
1126
1135
self.schema_jsons.push(resolve_schema_json(py, schema)?);
1127
1136
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1177,6 +1186,7 @@ impl JustAPIApp {
1177
1186
self.handlers.push(handler);
1178
1187
self.native.push(native.unwrap_or(false));
1179
1188
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1189
+ self.sse_specs.push(None);
1180
1190
self.schemas.push(None);
1181
1191
self.schema_jsons.push(None);
1182
1192
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1239,6 +1249,7 @@ impl JustAPIApp {
1239
1249
self.handlers.push(handler);
1240
1250
self.native.push(native.unwrap_or(false));
1241
1251
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1252
+ self.sse_specs.push(None);
1242
1253
self.schemas.push(body_schema.as_ref().map(|b| b.clone_ref(py)));
1243
1254
self.schema_jsons.push(resolve_schema_json(py, schema)?);
1244
1255
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1268,6 +1279,70 @@ impl JustAPIApp {
1268
1279
.map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))
1269
1280
}
1270
1281
1282
+ /// Register a Rust-native SSE stream route (ADR-088): the server streams
1283
+ /// `count` events at `interval_ms` spacing, generated entirely in Rust
1284
+ /// (tokio + mpsc + streaming_response). No Python handler, no GIL, no
1285
+ /// pump — the streaming runs at native speed.
1286
+ #[allow(clippy::too_many_arguments)]
1287
+ #[pyo3(signature = (path, count=10, interval_ms=100, tags=None, summary=None, description=None, deprecated=None, name=None, include_in_schema=None))]
1288
+ fn sse_native(
1289
+ &mut self,
1290
+ py: Python<'_>,
1291
+ path: &str,
1292
+ count: u64,
1293
+ interval_ms: u64,
1294
+ tags: Option<Vec<String>>,
1295
+ summary: Option<String>,
1296
+ description: Option<String>,
1297
+ deprecated: Option<bool>,
1298
+ name: Option<String>,
1299
+ include_in_schema: Option<bool>,
1300
+ ) -> PyResult<()> {
1301
+ let key = (Method::GET, path.to_string());
1302
+ if let Some(ref n) = name {
1303
+ self.named_routes.insert(n.clone(), path.to_string());
1304
+ }
1305
+ // Placeholder handler — never invoked; the dispatch recognizes the
1306
+ // sse_spec and serves the Rust stream directly.
1307
+ let handler = py.None();
1308
+ if let Some(&id) = self.route_indices.get(&key) {
1309
+ self.handlers[id] = handler;
1310
+ self.sse_specs[id] = Some((count, interval_ms));
1311
+ } else {
1312
+ let id = self.handlers.len();
1313
+ self.handlers.push(handler);
1314
+ self.native.push(false);
1315
+ self.crud.push(None);
1316
+ self.sse_specs.push(Some((count, interval_ms)));
1317
+ self.schemas.push(None);
1318
+ self.schema_jsons.push(None);
1319
+ self.query_schema_jsons.push(None);
1320
+ self.batch_configs.push(None);
1321
+ self.route_indices.insert(key, id);
1322
+ self.router
1323
+ .insert(Method::GET, path, id)
1324
+ .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
1325
+ }
1326
+ self.store_meta(
1327
+ py,
1328
+ "GET".to_string(),
1329
+ path.to_string(),
1330
+ None,
1331
+ None,
1332
+ tags,
1333
+ summary,
1334
+ description,
1335
+ deprecated,
1336
+ None,
1337
+ None,
1338
+ None,
1339
+ None,
1340
+ false,
1341
+ Some(include_in_schema.unwrap_or(true)),
1342
+ );
1343
+ Ok(())
1344
+ }
1345
+
1271
1346
#[allow(clippy::too_many_arguments)]
1272
1347
#[pyo3(signature = (path, handler, query_schema=None, tags=None, summary=None, description=None, deprecated=None, status_code=None, responses=None, operation_id=None, openapi_extra=None, include_in_schema=None, name=None, native=None, crud_table=None, crud_columns=None))]
1273
1348
fn head(
@@ -1294,6 +1369,7 @@ impl JustAPIApp {
1294
1369
self.handlers.push(handler);
1295
1370
self.native.push(native.unwrap_or(false));
1296
1371
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1372
+ self.sse_specs.push(None);
1297
1373
self.schemas.push(None);
1298
1374
self.schema_jsons.push(None);
1299
1375
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1349,6 +1425,7 @@ impl JustAPIApp {
1349
1425
self.handlers.push(handler);
1350
1426
self.native.push(native.unwrap_or(false));
1351
1427
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1428
+ self.sse_specs.push(None);
1352
1429
self.schemas.push(None);
1353
1430
self.schema_jsons.push(None);
1354
1431
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1404,6 +1481,7 @@ impl JustAPIApp {
1404
1481
self.handlers.push(handler);
1405
1482
self.native.push(native.unwrap_or(false));
1406
1483
self.crud.push(make_crud_spec(crud_table, crud_columns)?);
1484
+ self.sse_specs.push(None);
1407
1485
self.schemas.push(None);
1408
1486
self.schema_jsons.push(None);
1409
1487
self.query_schema_jsons.push(resolve_schema_json(py, query_schema)?);
@@ -1749,6 +1827,18 @@ impl JustAPIApp {
1749
1827
})
1750
1828
.collect();
1751
1829
let needs_request = Arc::new(needs_request);
1830
+ // Per-handler flag: `@native_async` — async handler whose framework
1831
+ // ops run in Rust; dispatch via the fastest async path (ADR-089).
1832
+ let native_async: Vec<bool> = handlers
1833
+ .iter()
1834
+ .map(|h| {
1835
+ h.bind(py)
1836
+ .getattr("_is_native_async")
1837
+ .and_then(|v| v.extract::<bool>())
1838
+ .unwrap_or(false)
1839
+ })
1840
+ .collect();
1841
+ let native_async = Arc::new(native_async);
1752
1842
// Per-handler flag: is this route served by the native Rust fast path
1753
1843
// (schema-backed, response = validated request body, no Python call)?
1754
1844
let native: Vec<bool> = std::mem::take(&mut app.native);
@@ -1759,6 +1849,7 @@ impl JustAPIApp {
1759
1849
// resolved at request time from `db_pool`.
1760
1850
let crud_specs: crate::native::handlers::CrudConfig =
1761
1851
Arc::new(std::mem::take(&mut app.crud));
1852
+ let sse_specs: Arc<Vec<Option<(u64, u64)>>> = Arc::new(std::mem::take(&mut app.sse_specs));
1762
1853
let schemas = Arc::new(std::mem::take(&mut app.schemas));
1763
1854
let schema_jsons = Arc::new(std::mem::take(&mut app.schema_jsons));
1764
1855
let query_schema_jsons = Arc::new(std::mem::take(&mut app.query_schema_jsons));
@@ -1984,7 +2075,9 @@ impl JustAPIApp {
1984
2075
app_py_http1,
1985
2076
needs_request.clone(),
1986
2077
native.clone(),
2078
+ native_async.clone(),
1987
2079
crud_specs.clone(),
2080
+ sse_specs.clone(),
1988
2081
schema_validators.clone(),
1989
2082
max_body_size,
1990
2083
);
@@ -2008,7 +2101,9 @@ impl JustAPIApp {
2008
2101
Python::attach(|py| app_py_http3.as_ref().as_ref().map(|a| a.clone_ref(py)));
2009
2102
let h3_needs = needs_request.clone();
2010
2103
let h3_native = native.clone();
2104
+ let h3_native_async = native_async.clone();
2011
2105
let h3_crud = crud_specs.clone();
2106
+ let h3_sse = sse_specs.clone();
2012
2107
let h3_validators = schema_validators.clone();
2013
2108
let h3_handler: justapi_core::middleware::HandlerFn<
2014
2109
http_body_util::Full<bytes::Bytes>,
@@ -2025,7 +2120,9 @@ impl JustAPIApp {
2025
2120
h3_app,
2026
2121
h3_needs,
2027
2122
h3_native,
2123
+ h3_native_async,
2028
2124
h3_crud,
2125
+ h3_sse,
2029
2126
h3_validators,
2030
2127
max_body_size,
2031
2128
);
2032
2129