Update pyo3 to 0.25 and convert to pyo3-async-runtimes (#897)

* Update pyo3 to 0.25 and convert to pyo3-async-runtimes

* Format and clippy

* Simplify awaitable conversion

Author: tconley1428

temporalio/bridge/Cargo.lock

@@ -21,9 +21,9 @@ async-trait = "0.1"
 futures = "0.3"
 log = "0.4"
 prost = "0.13"
-pyo3 = { version = "0.20", features = ["extension-module", "abi3-py39", "anyhow"] }
-pyo3-asyncio = { version = "0.20", features = ["tokio-runtime"] }
-pythonize = "0.20"
+pyo3 = { version = "0.25", features = ["extension-module", "abi3-py39", "anyhow"] }
+pyo3-async-runtimes = { version = "0.25", features = ["tokio-runtime"] }
+pythonize = "0.25"
 temporal-client = { version = "0.1.0", path = "./sdk-core/client" }
 temporal-sdk-core = { version = "0.1.0", path = "./sdk-core/core", features = ["ephemeral-server"] }
 temporal-sdk-core-api = { version = "0.1.0", path = "./sdk-core/core-api" }

temporalio/bridge/Cargo.toml

@@ -80,7 +80,7 @@ pub fn connect_client<'a>(
     py: Python<'a>,
     runtime_ref: &runtime::RuntimeRef,
     config: ClientConfig,
-) -> PyResult<&'a PyAny> {
+) -> PyResult<Bound<'a, PyAny>> {
     let opts: ClientOptions = config.try_into()?;
     let runtime = runtime_ref.runtime.clone();
     runtime_ref.runtime.future_into_py(py, async move {
@@ -126,7 +126,11 @@ impl ClientRef {
         self.retry_client.get_client().set_api_key(api_key);
     }
 
-    fn call_workflow_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<&'p PyAny> {
+    fn call_workflow_service<'p>(
+        &self,
+        py: Python<'p>,
+        call: RpcCall,
+    ) -> PyResult<Bound<'p, PyAny>> {
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match call.rpc.as_str() {
@@ -361,12 +365,15 @@ impl ClientRef {
                     )))
                 }
             }?;
-            let bytes: &[u8] = &bytes;
-            Ok(Python::with_gil(|py| bytes.into_py(py)))
+            Ok(bytes)
         })
     }
 
-    fn call_operator_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<&'p PyAny> {
+    fn call_operator_service<'p>(
+        &self,
+        py: Python<'p>,
+        call: RpcCall,
+    ) -> PyResult<Bound<'p, PyAny>> {
         use temporal_client::OperatorService;
 
         let mut retry_client = self.retry_client.clone();
@@ -403,12 +410,11 @@ impl ClientRef {
                     )))
                 }
             }?;
-            let bytes: &[u8] = &bytes;
-            Ok(Python::with_gil(|py| bytes.into_py(py)))
+            Ok(bytes)
         })
     }
 
-    fn call_cloud_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<&'p PyAny> {
+    fn call_cloud_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
         use temporal_client::CloudService;
 
         let mut retry_client = self.retry_client.clone();
@@ -466,12 +472,11 @@ impl ClientRef {
                     )))
                 }
             }?;
-            let bytes: &[u8] = &bytes;
-            Ok(Python::with_gil(|py| bytes.into_py(py)))
+            Ok(bytes)
         })
     }
 
-    fn call_test_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<&'p PyAny> {
+    fn call_test_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match call.rpc.as_str() {
@@ -490,12 +495,11 @@ impl ClientRef {
                     )))
                 }
             }?;
-            let bytes: &[u8] = &bytes;
-            Ok(Python::with_gil(|py| bytes.into_py(py)))
+            Ok(bytes)
         })
     }
 
-    fn call_health_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<&'p PyAny> {
+    fn call_health_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match call.rpc.as_str() {
@@ -507,8 +511,7 @@ impl ClientRef {
                     )))
                 }
             }?;
-            let bytes: &[u8] = &bytes;
-            Ok(Python::with_gil(|py| bytes.into_py(py)))
+            Ok(bytes)
         })
     }
 }
@@ -539,13 +542,13 @@ where
     match res {
         Ok(resp) => Ok(resp.get_ref().encode_to_vec()),
         Err(err) => {
-            Err(Python::with_gil(move |py| {
+            Python::with_gil(move |py| {
                 // Create tuple of "status", "message", and optional "details"
                 let code = err.code() as u32;
                 let message = err.message().to_owned();
-                let details = err.details().into_py(py);
-                RPCError::new_err((code, message, details))
-            }))
+                let details = err.details().into_pyobject(py)?.unbind();
+                Err(RPCError::new_err((code, message, details)))
+            })
         }
     }
 }

temporalio/bridge/src/client.rs

@@ -8,7 +8,7 @@ mod testing;
 mod worker;
 
 #[pymodule]
-fn temporal_sdk_bridge(py: Python, m: &PyModule) -> PyResult<()> {
+fn temporal_sdk_bridge(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     // Client stuff
     m.add("RPCError", py.get_type::<client::RPCError>())?;
     m.add_class::<client::ClientRef>()?;
@@ -62,7 +62,7 @@ fn connect_client<'a>(
     py: Python<'a>,
     runtime_ref: &runtime::RuntimeRef,
     config: client::ClientConfig,
-) -> PyResult<&'a PyAny> {
+) -> PyResult<Bound<'a, PyAny>> {
     client::connect_client(py, runtime_ref, config)
 }
 
@@ -77,7 +77,7 @@ fn init_runtime(telemetry_config: runtime::TelemetryConfig) -> PyResult<runtime:
 }
 
 #[pyfunction]
-fn raise_in_thread(py: Python, thread_id: std::os::raw::c_long, exc: &PyAny) -> bool {
+fn raise_in_thread(py: Python, thread_id: std::os::raw::c_long, exc: &Bound<'_, PyAny>) -> bool {
     runtime::raise_in_thread(py, thread_id, exc)
 }
 
@@ -86,7 +86,7 @@ fn start_dev_server<'a>(
     py: Python<'a>,
     runtime_ref: &runtime::RuntimeRef,
     config: testing::DevServerConfig,
-) -> PyResult<&'a PyAny> {
+) -> PyResult<Bound<'a, PyAny>> {
     testing::start_dev_server(py, runtime_ref, config)
 }
 
@@ -95,7 +95,7 @@ fn start_test_server<'a>(
     py: Python<'a>,
     runtime_ref: &runtime::RuntimeRef,
     config: testing::TestServerConfig,
-) -> PyResult<&'a PyAny> {
+) -> PyResult<Bound<'a, PyAny>> {
     testing::start_test_server(py, runtime_ref, config)
 }
 
@@ -113,6 +113,6 @@ fn new_replay_worker<'a>(
     py: Python<'a>,
     runtime_ref: &runtime::RuntimeRef,
     config: worker::WorkerConfig,
-) -> PyResult<&'a PyTuple> {
+) -> PyResult<Bound<'a, PyTuple>> {
     worker::new_replay_worker(py, runtime_ref, config)
 }