diff --git a/packages/pynumaflow-lite/Cargo.toml b/packages/pynumaflow-lite/Cargo.toml index 50824570..c1e6809f 100644 --- a/packages/pynumaflow-lite/Cargo.toml +++ b/packages/pynumaflow-lite/Cargo.toml @@ -10,7 +10,7 @@ crate-type = ["cdylib", "rlib"] [dependencies] numaflow = { git = "https://github.com/numaproj/numaflow-rs.git", rev = "fde3deafea634abbc347032ff409d33d4e1514b1" } -pyo3 = { version = "0.26.0", features = ["chrono", "experimental-inspect"] } +pyo3 = { version = "0.27.1", features = ["chrono", "experimental-inspect"] } tokio = "1.47.1" tonic = "0.14.2" tokio-stream = "0.1.17" @@ -18,7 +18,7 @@ tower = "0.5.2" hyper-util = "0.1.16" prost-types = "0.14.1" chrono = "0.4.42" -pyo3-async-runtimes = { version = "0.26.0", features = ["tokio-runtime"] } +pyo3-async-runtimes = { version = "0.27.0", features = ["tokio-runtime"] } futures-core = "0.3.31" pin-project = "1.1.10" diff --git a/packages/pynumaflow-lite/src/accumulate/server.rs b/packages/pynumaflow-lite/src/accumulate/server.rs index 842cb682..7a39054c 100644 --- a/packages/pynumaflow-lite/src/accumulate/server.rs +++ b/packages/pynumaflow-lite/src/accumulate/server.rs @@ -33,7 +33,7 @@ impl accumulator::AccumulatorCreator for PyAccumulatorCreator { match &self.init_args { Some(args) => { let bound = args.as_ref().bind(py); - let py_tuple = bound.downcast::()?; + let py_tuple = bound.cast::()?; class.call1(py, py_tuple) } None => class.call0(py), @@ -82,7 +82,7 @@ impl accumulator::Accumulator for PyAccumulatorRunner { .expect("python handler method raised before returning async iterable"); // Keep as Py - agen.extract(py).unwrap_or(agen) + agen.clone_ref(py).extract(py).unwrap_or(agen) }); // Wrap the Python AsyncIterable in a Rust Stream that yields incrementally diff --git a/packages/pynumaflow-lite/src/lib.rs b/packages/pynumaflow-lite/src/lib.rs index 200b3813..1522ae8e 100644 --- a/packages/pynumaflow-lite/src/lib.rs +++ b/packages/pynumaflow-lite/src/lib.rs @@ -82,7 +82,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.mapper` as well as attribute access let binding = m.getattr("mapper")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.mapper"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -91,7 +91,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.batchmapper` as well let binding = m.getattr("batchmapper")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.batchmapper"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -100,7 +100,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.mapstreamer` as well let binding = m.getattr("mapstreamer")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.mapstreamer"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -109,7 +109,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.reducer` as well let binding = m.getattr("reducer")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.reducer"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -118,7 +118,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.session_reducer` as well let binding = m.getattr("session_reducer")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.session_reducer"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -127,7 +127,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.accumulator` as well let binding = m.getattr("accumulator")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.accumulator"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -136,7 +136,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.sinker` as well let binding = m.getattr("sinker")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.sinker"; sub.setattr("__name__", fullname)?; py.import("sys")? @@ -145,7 +145,7 @@ fn pynumaflow_lite(py: Python, m: &Bound) -> PyResult<()> { // Ensure it's importable as `pynumaflow_lite.sourcer` as well let binding = m.getattr("sourcer")?; - let sub = binding.downcast::()?; + let sub = binding.cast::()?; let fullname = "pynumaflow_lite.sourcer"; sub.setattr("__name__", fullname)?; py.import("sys")? diff --git a/packages/pynumaflow-lite/src/mapstream/server.rs b/packages/pynumaflow-lite/src/mapstream/server.rs index 19f25cab..a90eda27 100644 --- a/packages/pynumaflow-lite/src/mapstream/server.rs +++ b/packages/pynumaflow-lite/src/mapstream/server.rs @@ -27,7 +27,7 @@ impl mapstream::MapStreamer for PyMapStreamRunner { .call1(py, (keys, datum)) .expect("python handler raised before returning async iterable"); // Keep as Py - agen.extract(py).unwrap_or(agen) + agen.clone_ref(py).extract(py).unwrap_or(agen) }); // Wrap the Python AsyncIterable in a Rust Stream that yields incrementally diff --git a/packages/pynumaflow-lite/src/pyiterables.rs b/packages/pynumaflow-lite/src/pyiterables.rs index 197b0e28..fc4e33c5 100644 --- a/packages/pynumaflow-lite/src/pyiterables.rs +++ b/packages/pynumaflow-lite/src/pyiterables.rs @@ -31,7 +31,7 @@ pub struct PyAsyncIterStream { impl PyAsyncIterStream where - M: for<'py> FromPyObject<'py> + Send + 'static, + M: for<'a, 'py> FromPyObject<'a, 'py> + Send + 'static, { /// Given a Python AsyncIterator and the event loop, build a stream over its items. /// It calls `__aiter__` on the `async_iterable` to get the iterator. @@ -48,7 +48,7 @@ where impl Stream for PyAsyncIterStream where - M: for<'py> FromPyObject<'py> + Send + 'static, + M: for<'a, 'py> FromPyObject<'a, 'py> + Send + 'static, { type Item = PyResult; @@ -117,7 +117,7 @@ where match res { Ok(obj) => { // Convert PyObject -> M - let m = Python::attach(|py| obj.extract::(py)); + let m = Python::attach(|py| obj.extract::(py).map_err(Into::into)); Poll::Ready(Some(m)) } Err(err) => { diff --git a/packages/pynumaflow-lite/src/reduce/server.rs b/packages/pynumaflow-lite/src/reduce/server.rs index 813af1aa..dc19de02 100644 --- a/packages/pynumaflow-lite/src/reduce/server.rs +++ b/packages/pynumaflow-lite/src/reduce/server.rs @@ -43,7 +43,7 @@ impl reduce::ReducerCreator for PyReduceCreator { match &self.init_args { Some(args) => { let bound = args.as_ref().bind(py); - let py_tuple = bound.downcast::()?; + let py_tuple = bound.cast::()?; class.call1(py, py_tuple) } None => class.call0(py), diff --git a/packages/pynumaflow-lite/src/session_reduce/server.rs b/packages/pynumaflow-lite/src/session_reduce/server.rs index d9006d63..51438eae 100644 --- a/packages/pynumaflow-lite/src/session_reduce/server.rs +++ b/packages/pynumaflow-lite/src/session_reduce/server.rs @@ -33,7 +33,7 @@ impl session_reduce::SessionReducerCreator for PySessionReduceCreator { match &self.init_args { Some(args) => { let bound = args.as_ref().bind(py); - let py_tuple = bound.downcast::()?; + let py_tuple = bound.cast::()?; class.call1(py, py_tuple) } None => class.call0(py), @@ -83,7 +83,7 @@ impl session_reduce::SessionReducer for PySessionReduceRunner { .expect("python session_reduce method raised before returning async iterable"); // Keep as Py - agen.extract(py).unwrap_or(agen) + agen.clone_ref(py).extract(py).unwrap_or(agen) }); // Wrap the Python AsyncIterable in a Rust Stream that yields incrementally