chore: upgrade pyo3 to 0.27 (from 0.26) (#286)

Signed-off-by: Vigith Maurice <[email protected]>

Author: vigith

packages/pynumaflow-lite/Cargo.toml

@@ -10,15 +10,15 @@ 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"
 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"
 

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::<PyTuple>()?;
+                    let py_tuple = bound.cast::<PyTuple>()?;
                     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<PyAny>
-            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

packages/pynumaflow-lite/src/lib.rs

@@ -82,7 +82,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.mapper` as well as attribute access
     let binding = m.getattr("mapper")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.mapper";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -91,7 +91,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.batchmapper` as well
     let binding = m.getattr("batchmapper")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.batchmapper";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -100,7 +100,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.mapstreamer` as well
     let binding = m.getattr("mapstreamer")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.mapstreamer";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -109,7 +109,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.reducer` as well
     let binding = m.getattr("reducer")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.reducer";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -118,7 +118,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.session_reducer` as well
     let binding = m.getattr("session_reducer")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.session_reducer";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -127,7 +127,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.accumulator` as well
     let binding = m.getattr("accumulator")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.accumulator";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -136,7 +136,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.sinker` as well
     let binding = m.getattr("sinker")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.sinker";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?
@@ -145,7 +145,7 @@ fn pynumaflow_lite(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
 
     // Ensure it's importable as `pynumaflow_lite.sourcer` as well
     let binding = m.getattr("sourcer")?;
-    let sub = binding.downcast::<PyModule>()?;
+    let sub = binding.cast::<PyModule>()?;
     let fullname = "pynumaflow_lite.sourcer";
     sub.setattr("__name__", fullname)?;
     py.import("sys")?

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<PyAny>
-            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

packages/pynumaflow-lite/src/pyiterables.rs

@@ -31,7 +31,7 @@ pub struct PyAsyncIterStream<M> {
 
 impl<M> PyAsyncIterStream<M>
 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<M> Stream for PyAsyncIterStream<M>
 where
-    M: for<'py> FromPyObject<'py> + Send + 'static,
+    M: for<'a, 'py> FromPyObject<'a, 'py> + Send + 'static,
 {
     type Item = PyResult<M>;
 
@@ -117,7 +117,7 @@ where
                     match res {
                         Ok(obj) => {
                             // Convert PyObject -> M
-                            let m = Python::attach(|py| obj.extract::<M>(py));
+                            let m = Python::attach(|py| obj.extract::<M>(py).map_err(Into::into));
                             Poll::Ready(Some(m))
                         }
                         Err(err) => {

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::<PyTuple>()?;
+                        let py_tuple = bound.cast::<PyTuple>()?;
                         class.call1(py, py_tuple)
                     }
                     None => class.call0(py),

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::<PyTuple>()?;
+                    let py_tuple = bound.cast::<PyTuple>()?;
                     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<PyAny>
-            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