add python bridge

Signed-off-by: michaelfeil <[email protected]>

Author: michaelfeil

lib/bindings/python/Cargo.lock

@@ -47,6 +47,7 @@ tokio-util = { version = "0.7", features = ["rt"] }
 tracing = { version = "0"  }
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
 uuid = { version = "1.17", features = ["v4", "serde"] }
+crossbeam-channel = "0.5"
 
 # "extension-module" tells pyo3 we want to build an extension module (skips linking against libpython.so)
 # "abi3-py310" tells pyo3 (and maturin) to build using the stable ABI with minimum Python version 3.10, which is the minimum version in pyproject.toml

lib/bindings/python/Cargo.toml

@@ -1,6 +1,6 @@
 use std::{
     sync::{OnceLock, mpsc},
-    thread::{self, ThreadId},
+    thread::{self},
 };
 
 use crossbeam_channel::{Sender, bounded};
@@ -135,4 +135,4 @@ impl Bridge {
             .recv()
             .map_err(|_| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>("bridge dropped"))?
     }
-}
+}

lib/bindings/python/rust/bridge.rs

@@ -8,19 +8,19 @@ use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyModule};
 use pyo3::{PyAny, PyErr};
 use pyo3_async_runtimes::TaskLocals;
-use pythonize::{depythonize, pythonize};
+use pythonize::pythonize;
 pub use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 use tokio_util::sync::CancellationToken;
 
+use super::context::{Context, callable_accepts_kwarg};
+use crate::bridge;
 use dynamo_runtime::logging::get_distributed_tracing_context;
 pub use dynamo_runtime::{
     pipeline::{AsyncEngine, AsyncEngineContextProvider, Data, ManyOut, ResponseStream, SingleIn},
     protocols::annotated::Annotated,
 };
-use crate::bridge;
-use super::context::{Context, callable_accepts_kwarg};
 
 /// Add bingings from this crate to the provided module
 pub fn add_to_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
@@ -182,7 +182,6 @@ where
         let stream = bridge::Bridge::global()
             .with_gil(move |py| {
                 let py_request = pythonize(py, &request)?;
-                let id = ctx_python.id().to_string();
 
                 let py_ctx = Py::new(py, Context::new(ctx_python.clone(), current_trace_context))?;
 

lib/bindings/python/rust/engine.rs

@@ -57,6 +57,7 @@ impl From<RouterMode> for RsRouterMode {
     }
 }
 
+mod bridge;
 mod context;
 mod engine;
 mod http;
@@ -66,6 +67,8 @@ mod parsers;
 mod planner;
 mod prometheus_metrics;
 
+use bridge::Bridge;
+
 type JsonServerStreamingIngress =
     Ingress<SingleIn<serde_json::Value>, ManyOut<RsAnnotated<serde_json::Value>>>;
 
@@ -899,6 +902,7 @@ impl Client {
         context: Option<context::Context>,
     ) -> PyResult<Bound<'p, PyAny>> {
         // Convert to an owned GIL-independent handle and move into async:
+        #[allow(deprecated)]
         let request_py: Py<PyAny> = request.into_py(py);
         let annotated = annotated.unwrap_or(false);
 
@@ -940,6 +944,7 @@ impl Client {
         context: Option<context::Context>,
     ) -> PyResult<Bound<'p, PyAny>> {
         // Convert to an owned GIL-independent handle and move into async:
+        #[allow(deprecated)]
         let request_py: Py<PyAny> = request.into_py(py);
         let annotated = annotated.unwrap_or(false);
 
@@ -982,6 +987,7 @@ impl Client {
         context: Option<context::Context>,
     ) -> PyResult<Bound<'p, PyAny>> {
         // Convert to an owned GIL-independent handle and move into async:
+        #[allow(deprecated)]
         let request_py: Py<PyAny> = request.into_py(py);
         let annotated = annotated.unwrap_or(false);
 
@@ -1028,13 +1034,14 @@ async fn process_stream(
     while let Some(response) = stream.next().await {
         // Convert the response to a PyObject using Python's GIL
         let annotated: RsAnnotated<serde_json::Value> = response;
-        let annotated: RsAnnotated<PyObject> = .map_data_async(|data| async move {
-            Bridge::global()
+        let annotated: RsAnnotated<PyObject> = annotated
+            .map_data_async(|data| async move {
+                Bridge::global()
                     .to_py(data) // runs on bridge worker thread
                     .await
                     .map_err(|e| e.to_string())
-                    .await;
-        });
+            })
+            .await;
 
         let is_error = annotated.is_error();