pyo3_async_runtimes/lib.rs
1#![warn(missing_docs)]
2#![allow(clippy::borrow_deref_ref)]
3
4//! Rust Bindings to the Python Asyncio Event Loop
5//!
6//! # Motivation
7//!
8//! This crate aims to provide a convenient interface to manage the interop between Python and
9//! Rust's async/await models. It supports conversions between Rust and Python futures and manages
10//! the event loops for both languages. Python's threading model and GIL can make this interop a bit
11//! trickier than one might expect, so there are a few caveats that users should be aware of.
12//!
13//! ## Why Two Event Loops
14//!
15//! Currently, we don't have a way to run Rust futures directly on Python's event loop. Likewise,
16//! Python's coroutines cannot be directly spawned on a Rust event loop. The two coroutine models
17//! require some additional assistance from their event loops, so in all likelihood they will need
18//! a new _unique_ event loop that addresses the needs of both languages if the coroutines are to
19//! be run on the same loop.
20//!
21//! It's not immediately clear that this would provide worthwhile performance wins either, so in the
22//! interest of getting something simple out there to facilitate these conversions, this crate
23//! handles the communication between _separate_ Python and Rust event loops.
24//!
25//! ## Python's Event Loop and the Main Thread
26//!
27//! Python is very picky about the threads used by the `asyncio` executor. In particular, it needs
28//! to have control over the main thread in order to handle signals like CTRL-C correctly. This
29//! means that Cargo's default test harness will no longer work since it doesn't provide a method of
30//! overriding the main function to add our event loop initialization and finalization.
31//!
32//! ## Event Loop References and ContextVars
33//!
34//! One problem that arises when interacting with Python's asyncio library is that the functions we
35//! use to get a reference to the Python event loop can only be called in certain contexts. Since
36//! PyO3 Asyncio needs to interact with Python's event loop during conversions, the context of these
37//! conversions can matter a lot.
38//!
39//! Likewise, Python's `contextvars` library can require some special treatment. Python functions
40//! and coroutines can rely on the context of outer coroutines to function correctly, so this
41//! library needs to be able to preserve `contextvars` during conversions.
42//!
43//! > The core conversions we've mentioned so far in the README should insulate you from these
44//! > concerns in most cases. For the edge cases where they don't, this section should provide you
45//! > with the information you need to solve these problems.
46//!
47//! ### The Main Dilemma
48//!
49//! Python programs can have many independent event loop instances throughout the lifetime of the
50//! application (`asyncio.run` for example creates its own event loop each time it's called for
51//! instance), and they can even run concurrent with other event loops. For this reason, the most
52//! correct method of obtaining a reference to the Python event loop is via
53//! `asyncio.get_running_loop`.
54//!
55//! `asyncio.get_running_loop` returns the event loop associated with the current OS thread. It can
56//! be used inside Python coroutines to spawn concurrent tasks, interact with timers, or in our case
57//! signal between Rust and Python. This is all well and good when we are operating on a Python
58//! thread, but since Rust threads are not associated with a Python event loop,
59//! `asyncio.get_running_loop` will fail when called on a Rust runtime.
60//!
61//! `contextvars` operates in a similar way, though the current context is not always associated
62//! with the current OS thread. Different contexts can be associated with different coroutines even
63//! if they run on the same OS thread.
64//!
65//! ### The Solution
66//!
67//! A really straightforward way of dealing with this problem is to pass references to the
68//! associated Python event loop and context for every conversion. That's why we have a structure
69//! called `TaskLocals` and a set of conversions that accept it.
70//!
71//! `TaskLocals` stores the current event loop, and allows the user to copy the current Python
72//! context if necessary. The following conversions will use these references to perform the
73//! necessary conversions and restore Python context when needed:
74//!
75//! - `pyo3_async_runtimes::into_future_with_locals` - Convert a Python awaitable into a Rust future.
76//! - `pyo3_async_runtimes::<runtime>::future_into_py_with_locals` - Convert a Rust future into a Python
77//! awaitable.
78//! - `pyo3_async_runtimes::<runtime>::local_future_into_py_with_locals` - Convert a `!Send` Rust future
79//! into a Python awaitable.
80//!
81//! One clear disadvantage to this approach is that the Rust application has to explicitly track
82//! these references. In native libraries, we can't make any assumptions about the underlying event
83//! loop, so the only reliable way to make sure our conversions work properly is to store these
84//! references at the callsite to use later on.
85//!
86//! ```rust
87//! use pyo3::{wrap_pyfunction, prelude::*};
88//!
89//! # #[cfg(feature = "tokio-runtime")]
90//! #[pyfunction]
91//! fn sleep(py: Python) -> PyResult<Bound<PyAny>> {
92//! // Construct the task locals structure with the current running loop and context
93//! let locals = pyo3_async_runtimes::TaskLocals::with_running_loop(py)?.copy_context(py)?;
94//!
95//! // Convert the async move { } block to a Python awaitable
96//! pyo3_async_runtimes::tokio::future_into_py_with_locals(py, locals.clone_ref(py), async move {
97//! let py_sleep = Python::with_gil(|py| {
98//! // Sometimes we need to call other async Python functions within
99//! // this future. In order for this to work, we need to track the
100//! // event loop from earlier.
101//! pyo3_async_runtimes::into_future_with_locals(
102//! &locals,
103//! py.import("asyncio")?.call_method1("sleep", (1,))?
104//! )
105//! })?;
106//!
107//! py_sleep.await?;
108//!
109//! Ok(())
110//! })
111//! }
112//!
113//! # #[cfg(feature = "tokio-runtime")]
114//! #[pymodule]
115//! fn my_mod(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
116//! m.add_function(wrap_pyfunction!(sleep, m)?)?;
117//! Ok(())
118//! }
119//! ```
120//!
121//! > A naive solution to this tracking problem would be to cache a global reference to the asyncio
122//! > event loop that all PyO3 Asyncio conversions can use. In fact this is what we did in PyO3
123//! > Asyncio `v0.13`. This works well for applications, but it soon became clear that this is not
124//! > so ideal for libraries. Libraries usually have no direct control over how the event loop is
125//! > managed, they're just expected to work with any event loop at any point in the application.
126//! > This problem is compounded further when multiple event loops are used in the application since
127//! > the global reference will only point to one.
128//!
129//! Another disadvantage to this explicit approach that is less obvious is that we can no longer
130//! call our `#[pyfunction] fn sleep` on a Rust runtime since `asyncio.get_running_loop` only works
131//! on Python threads! It's clear that we need a slightly more flexible approach.
132//!
133//! In order to detect the Python event loop at the callsite, we need something like
134//! `asyncio.get_running_loop` and `contextvars.copy_context` that works for _both Python and Rust_.
135//! In Python, `asyncio.get_running_loop` uses thread-local data to retrieve the event loop
136//! associated with the current thread. What we need in Rust is something that can retrieve the
137//! Python event loop and contextvars associated with the current Rust _task_.
138//!
139//! Enter `pyo3_async_runtimes::<runtime>::get_current_locals`. This function first checks task-local data
140//! for the `TaskLocals`, then falls back on `asyncio.get_running_loop` and
141//! `contextvars.copy_context` if no task locals are found. This way both bases are
142//! covered.
143//!
144//! Now, all we need is a way to store the `TaskLocals` for the Rust future. Since this is a
145//! runtime-specific feature, you can find the following functions in each runtime module:
146//!
147//! - `pyo3_async_runtimes::<runtime>::scope` - Store the task-local data when executing the given Future.
148//! - `pyo3_async_runtimes::<runtime>::scope_local` - Store the task-local data when executing the given
149//! `!Send` Future.
150//!
151//! With these new functions, we can make our previous example more correct:
152//!
153//! ```rust no_run
154//! use pyo3::prelude::*;
155//!
156//! # #[cfg(feature = "tokio-runtime")]
157//! #[pyfunction]
158//! fn sleep(py: Python) -> PyResult<Bound<PyAny>> {
159//! // get the current event loop through task-local data
160//! // OR `asyncio.get_running_loop` and `contextvars.copy_context`
161//! let locals = pyo3_async_runtimes::tokio::get_current_locals(py)?;
162//!
163//! pyo3_async_runtimes::tokio::future_into_py_with_locals(
164//! py,
165//! locals.clone_ref(py),
166//! // Store the current locals in task-local data
167//! pyo3_async_runtimes::tokio::scope(locals.clone_ref(py), async move {
168//! let py_sleep = Python::with_gil(|py| {
169//! pyo3_async_runtimes::into_future_with_locals(
170//! // Now we can get the current locals through task-local data
171//! &pyo3_async_runtimes::tokio::get_current_locals(py)?,
172//! py.import("asyncio")?.call_method1("sleep", (1,))?
173//! )
174//! })?;
175//!
176//! py_sleep.await?;
177//!
178//! Ok(Python::with_gil(|py| py.None()))
179//! })
180//! )
181//! }
182//!
183//! # #[cfg(feature = "tokio-runtime")]
184//! #[pyfunction]
185//! fn wrap_sleep(py: Python) -> PyResult<Bound<PyAny>> {
186//! // get the current event loop through task-local data
187//! // OR `asyncio.get_running_loop` and `contextvars.copy_context`
188//! let locals = pyo3_async_runtimes::tokio::get_current_locals(py)?;
189//!
190//! pyo3_async_runtimes::tokio::future_into_py_with_locals(
191//! py,
192//! locals.clone_ref(py),
193//! // Store the current locals in task-local data
194//! pyo3_async_runtimes::tokio::scope(locals.clone_ref(py), async move {
195//! let py_sleep = Python::with_gil(|py| {
196//! pyo3_async_runtimes::into_future_with_locals(
197//! &pyo3_async_runtimes::tokio::get_current_locals(py)?,
198//! // We can also call sleep within a Rust task since the
199//! // locals are stored in task local data
200//! sleep(py)?
201//! )
202//! })?;
203//!
204//! py_sleep.await?;
205//!
206//! Ok(Python::with_gil(|py| py.None()))
207//! })
208//! )
209//! }
210//!
211//! # #[cfg(feature = "tokio-runtime")]
212//! #[pymodule]
213//! fn my_mod(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
214//! m.add_function(wrap_pyfunction!(sleep, m)?)?;
215//! m.add_function(wrap_pyfunction!(wrap_sleep, m)?)?;
216//! Ok(())
217//! }
218//! ```
219//!
220//! Even though this is more correct, it's clearly not more ergonomic. That's why we introduced a
221//! set of functions with this functionality baked in:
222//!
223//! - `pyo3_async_runtimes::<runtime>::into_future`
224//! > Convert a Python awaitable into a Rust future (using
225//! > `pyo3_async_runtimes::<runtime>::get_current_locals`)
226//! - `pyo3_async_runtimes::<runtime>::future_into_py`
227//! > Convert a Rust future into a Python awaitable (using
228//! > `pyo3_async_runtimes::<runtime>::get_current_locals` and `pyo3_async_runtimes::<runtime>::scope` to set the
229//! > task-local event loop for the given Rust future)
230//! - `pyo3_async_runtimes::<runtime>::local_future_into_py`
231//! > Convert a `!Send` Rust future into a Python awaitable (using
232//! > `pyo3_async_runtimes::<runtime>::get_current_locals` and `pyo3_async_runtimes::<runtime>::scope_local` to
233//! > set the task-local event loop for the given Rust future).
234//!
235//! __These are the functions that we recommend using__. With these functions, the previous example
236//! can be rewritten to be more compact:
237//!
238//! ```rust
239//! use pyo3::prelude::*;
240//!
241//! # #[cfg(feature = "tokio-runtime")]
242//! #[pyfunction]
243//! fn sleep(py: Python) -> PyResult<Bound<PyAny>> {
244//! pyo3_async_runtimes::tokio::future_into_py(py, async move {
245//! let py_sleep = Python::with_gil(|py| {
246//! pyo3_async_runtimes::tokio::into_future(
247//! py.import("asyncio")?.call_method1("sleep", (1,))?
248//! )
249//! })?;
250//!
251//! py_sleep.await?;
252//!
253//! Ok(Python::with_gil(|py| py.None()))
254//! })
255//! }
256//!
257//! # #[cfg(feature = "tokio-runtime")]
258//! #[pyfunction]
259//! fn wrap_sleep(py: Python) -> PyResult<Bound<PyAny>> {
260//! pyo3_async_runtimes::tokio::future_into_py(py, async move {
261//! let py_sleep = Python::with_gil(|py| {
262//! pyo3_async_runtimes::tokio::into_future(sleep(py)?)
263//! })?;
264//!
265//! py_sleep.await?;
266//!
267//! Ok(Python::with_gil(|py| py.None()))
268//! })
269//! }
270//!
271//! # #[cfg(feature = "tokio-runtime")]
272//! #[pymodule]
273//! fn my_mod(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
274//! m.add_function(wrap_pyfunction!(sleep, m)?)?;
275//! m.add_function(wrap_pyfunction!(wrap_sleep, m)?)?;
276//! Ok(())
277//! }
278//! ```
279//!
280//! > A special thanks to [@ShadowJonathan](https://github.com/ShadowJonathan) for helping with the
281//! > design and review of these changes!
282//!
283//! ## Rust's Event Loop
284//!
285//! Currently only the Async-Std and Tokio runtimes are supported by this crate. If you need support
286//! for another runtime, feel free to make a request on GitHub (or attempt to add support yourself
287//! with the [`generic`] module)!
288//!
289//! > _In the future, we may implement first class support for more Rust runtimes. Contributions are
290//! > welcome as well!_
291//!
292//! ## Features
293//!
294//! Items marked with
295//! <span
296//! class="module-item stab portability"
297//! style="display: inline; border-radius: 3px; padding: 2px; font-size: 80%; line-height: 1.2;"
298//! ><code>attributes</code></span>
299//! > are only available when the `attributes` Cargo feature is enabled:
300//!
301//! ```toml
302//! [dependencies.pyo3-async-runtimes]
303//! version = "0.24"
304//! features = ["attributes"]
305//! ```
306//!
307//! Items marked with
308//! <span
309//! class="module-item stab portability"
310//! style="display: inline; border-radius: 3px; padding: 2px; font-size: 80%; line-height: 1.2;"
311//! ><code>async-std-runtime</code></span>
312//! > are only available when the `async-std-runtime` Cargo feature is enabled:
313//!
314//! ```toml
315//! [dependencies.pyo3-async-runtimes]
316//! version = "0.24"
317//! features = ["async-std-runtime"]
318//! ```
319//!
320//! Items marked with
321//! <span
322//! class="module-item stab portability"
323//! style="display: inline; border-radius: 3px; padding: 2px; font-size: 80%; line-height: 1.2;"
324//! ><code>tokio-runtime</code></span>
325//! > are only available when the `tokio-runtime` Cargo feature is enabled:
326//!
327//! ```toml
328//! [dependencies.pyo3-async-runtimes]
329//! version = "0.24"
330//! features = ["tokio-runtime"]
331//! ```
332//!
333//! Items marked with
334//! <span
335//! class="module-item stab portability"
336//! style="display: inline; border-radius: 3px; padding: 2px; font-size: 80%; line-height: 1.2;"
337//! ><code>testing</code></span>
338//! > are only available when the `testing` Cargo feature is enabled:
339//!
340//! ```toml
341//! [dependencies.pyo3-async-runtimes]
342//! version = "0.24"
343//! features = ["testing"]
344//! ```
345
346/// Re-exported for #[test] attributes
347#[cfg(all(feature = "attributes", feature = "testing"))]
348pub use inventory;
349
350/// <span class="module-item stab portability" style="display: inline; border-radius: 3px; padding: 2px; font-size: 80%; line-height: 1.2;"><code>testing</code></span> Utilities for writing PyO3 Asyncio tests
351#[cfg(feature = "testing")]
352pub mod testing;
353
354#[cfg(feature = "async-std")]
355pub mod async_std;
356
357#[cfg(feature = "tokio-runtime")]
358pub mod tokio;
359
360/// Errors and exceptions related to PyO3 Asyncio
361pub mod err;
362
363pub mod generic;
364
365#[pymodule]
366fn pyo3_async_runtimes(py: Python, m: &Bound<PyModule>) -> PyResult<()> {
367 m.add("RustPanic", py.get_type::<err::RustPanic>())?;
368 Ok(())
369}
370
371/// Test README
372#[doc(hidden)]
373pub mod doc_test {
374 #[allow(unused)]
375 macro_rules! doc_comment {
376 ($x:expr, $module:item) => {
377 #[doc = $x]
378 $module
379 };
380 }
381
382 #[allow(unused)]
383 macro_rules! doctest {
384 ($x:expr, $y:ident) => {
385 doc_comment!(include_str!($x), mod $y {});
386 };
387 }
388
389 #[cfg(all(
390 feature = "async-std-runtime",
391 feature = "tokio-runtime",
392 feature = "attributes"
393 ))]
394 doctest!("../README.md", readme_md);
395}
396
397use std::future::Future;
398
399use futures::channel::oneshot;
400use once_cell::sync::OnceCell;
401use pyo3::{call::PyCallArgs, prelude::*, types::PyDict};
402
403static ASYNCIO: OnceCell<PyObject> = OnceCell::new();
404static CONTEXTVARS: OnceCell<PyObject> = OnceCell::new();
405static ENSURE_FUTURE: OnceCell<PyObject> = OnceCell::new();
406static GET_RUNNING_LOOP: OnceCell<PyObject> = OnceCell::new();
407
408fn ensure_future<'p>(py: Python<'p>, awaitable: &Bound<'p, PyAny>) -> PyResult<Bound<'p, PyAny>> {
409 ENSURE_FUTURE
410 .get_or_try_init(|| -> PyResult<PyObject> {
411 Ok(asyncio(py)?.getattr("ensure_future")?.into())
412 })?
413 .bind(py)
414 .call1((awaitable,))
415}
416
417fn create_future(event_loop: Bound<PyAny>) -> PyResult<Bound<'_, PyAny>> {
418 event_loop.call_method0("create_future")
419}
420
421fn close(event_loop: Bound<PyAny>) -> PyResult<()> {
422 event_loop.call_method1(
423 "run_until_complete",
424 (event_loop.call_method0("shutdown_asyncgens")?,),
425 )?;
426
427 // how to do this prior to 3.9?
428 if event_loop.hasattr("shutdown_default_executor")? {
429 event_loop.call_method1(
430 "run_until_complete",
431 (event_loop.call_method0("shutdown_default_executor")?,),
432 )?;
433 }
434
435 event_loop.call_method0("close")?;
436
437 Ok(())
438}
439
440fn asyncio(py: Python) -> PyResult<&Bound<PyAny>> {
441 ASYNCIO
442 .get_or_try_init(|| Ok(py.import("asyncio")?.into()))
443 .map(|asyncio| asyncio.bind(py))
444}
445
446/// Get a reference to the Python Event Loop from Rust
447///
448/// Equivalent to `asyncio.get_running_loop()` in Python 3.7+.
449pub fn get_running_loop(py: Python) -> PyResult<Bound<PyAny>> {
450 // Ideally should call get_running_loop, but calls get_event_loop for compatibility when
451 // get_running_loop is not available.
452 GET_RUNNING_LOOP
453 .get_or_try_init(|| -> PyResult<PyObject> {
454 let asyncio = asyncio(py)?;
455
456 Ok(asyncio.getattr("get_running_loop")?.into())
457 })?
458 .bind(py)
459 .call0()
460}
461
462fn contextvars(py: Python) -> PyResult<&Bound<PyAny>> {
463 Ok(CONTEXTVARS
464 .get_or_try_init(|| py.import("contextvars").map(|m| m.into()))?
465 .bind(py))
466}
467
468fn copy_context(py: Python) -> PyResult<Bound<PyAny>> {
469 contextvars(py)?.call_method0("copy_context")
470}
471
472/// Task-local data to store for Python conversions.
473#[derive(Debug)]
474pub struct TaskLocals {
475 /// Track the event loop of the Python task
476 event_loop: PyObject,
477 /// Track the contextvars of the Python task
478 context: PyObject,
479}
480
481impl TaskLocals {
482 /// At a minimum, TaskLocals must store the event loop.
483 pub fn new(event_loop: Bound<PyAny>) -> Self {
484 Self {
485 context: event_loop.py().None(),
486 event_loop: event_loop.into(),
487 }
488 }
489
490 /// Construct TaskLocals with the event loop returned by `get_running_loop`
491 pub fn with_running_loop(py: Python) -> PyResult<Self> {
492 Ok(Self::new(get_running_loop(py)?))
493 }
494
495 /// Manually provide the contextvars for the current task.
496 pub fn with_context(self, context: Bound<PyAny>) -> Self {
497 Self {
498 context: context.into(),
499 ..self
500 }
501 }
502
503 /// Capture the current task's contextvars
504 pub fn copy_context(self, py: Python) -> PyResult<Self> {
505 Ok(self.with_context(copy_context(py)?))
506 }
507
508 /// Get a reference to the event loop
509 pub fn event_loop<'p>(&self, py: Python<'p>) -> Bound<'p, PyAny> {
510 self.event_loop.clone_ref(py).into_bound(py)
511 }
512
513 /// Get a reference to the python context
514 pub fn context<'p>(&self, py: Python<'p>) -> Bound<'p, PyAny> {
515 self.context.clone_ref(py).into_bound(py)
516 }
517
518 /// Create a clone of the TaskLocals by incrementing the reference counters of the event loop and
519 /// contextvars.
520 pub fn clone_ref(&self, py: Python<'_>) -> Self {
521 Self {
522 event_loop: self.event_loop.clone_ref(py),
523 context: self.context.clone_ref(py),
524 }
525 }
526}
527
528#[pyclass]
529struct PyTaskCompleter {
530 tx: Option<oneshot::Sender<PyResult<PyObject>>>,
531}
532
533#[pymethods]
534impl PyTaskCompleter {
535 #[pyo3(signature = (task))]
536 pub fn __call__(&mut self, task: &Bound<PyAny>) -> PyResult<()> {
537 debug_assert!(task.call_method0("done")?.extract()?);
538 let result = match task.call_method0("result") {
539 Ok(val) => Ok(val.into()),
540 Err(e) => Err(e),
541 };
542
543 // unclear to me whether or not this should be a panic or silent error.
544 //
545 // calling PyTaskCompleter twice should not be possible, but I don't think it really hurts
546 // anything if it happens.
547 if let Some(tx) = self.tx.take() {
548 if tx.send(result).is_err() {
549 // cancellation is not an error
550 }
551 }
552
553 Ok(())
554 }
555}
556
557#[pyclass]
558struct PyEnsureFuture {
559 awaitable: PyObject,
560 tx: Option<oneshot::Sender<PyResult<PyObject>>>,
561}
562
563#[pymethods]
564impl PyEnsureFuture {
565 pub fn __call__(&mut self) -> PyResult<()> {
566 Python::with_gil(|py| {
567 let task = ensure_future(py, self.awaitable.bind(py))?;
568 let on_complete = PyTaskCompleter { tx: self.tx.take() };
569 task.call_method1("add_done_callback", (on_complete,))?;
570
571 Ok(())
572 })
573 }
574}
575
576fn call_soon_threadsafe<'py>(
577 event_loop: &Bound<'py, PyAny>,
578 context: &Bound<PyAny>,
579 args: impl PyCallArgs<'py>,
580) -> PyResult<()> {
581 let py = event_loop.py();
582
583 let kwargs = PyDict::new(py);
584 kwargs.set_item("context", context)?;
585
586 event_loop.call_method("call_soon_threadsafe", args, Some(&kwargs))?;
587 Ok(())
588}
589
590/// Convert a Python `awaitable` into a Rust Future
591///
592/// This function converts the `awaitable` into a Python Task using `run_coroutine_threadsafe`. A
593/// completion handler sends the result of this Task through a
594/// `futures::channel::oneshot::Sender<PyResult<PyObject>>` and the future returned by this function
595/// simply awaits the result through the `futures::channel::oneshot::Receiver<PyResult<PyObject>>`.
596///
597/// # Arguments
598/// * `locals` - The Python event loop and context to be used for the provided awaitable
599/// * `awaitable` - The Python `awaitable` to be converted
600///
601/// # Examples
602///
603/// ```
604/// use std::time::Duration;
605/// use std::ffi::CString;
606///
607/// use pyo3::prelude::*;
608///
609/// const PYTHON_CODE: &'static str = r#"
610/// import asyncio
611///
612/// async def py_sleep(duration):
613/// await asyncio.sleep(duration)
614/// "#;
615///
616/// # #[cfg(feature = "tokio-runtime")]
617/// async fn py_sleep(seconds: f32) -> PyResult<()> {
618/// let test_mod = Python::with_gil(|py| -> PyResult<PyObject> {
619/// Ok(
620/// PyModule::from_code(
621/// py,
622/// &CString::new(PYTHON_CODE).unwrap(),
623/// &CString::new("test_into_future/test_mod.py").unwrap(),
624/// &CString::new("test_mod").unwrap(),
625/// )?
626/// .into()
627/// )
628/// })?;
629///
630/// Python::with_gil(|py| {
631/// pyo3_async_runtimes::into_future_with_locals(
632/// &pyo3_async_runtimes::tokio::get_current_locals(py)?,
633/// test_mod
634/// .call_method1(py, "py_sleep", (seconds.into_pyobject(py).unwrap(),))?
635/// .into_bound(py),
636/// )
637/// })?
638/// .await?;
639/// Ok(())
640/// }
641/// ```
642pub fn into_future_with_locals(
643 locals: &TaskLocals,
644 awaitable: Bound<PyAny>,
645) -> PyResult<impl Future<Output = PyResult<PyObject>> + Send> {
646 let py = awaitable.py();
647 let (tx, rx) = oneshot::channel();
648
649 call_soon_threadsafe(
650 &locals.event_loop(py),
651 &locals.context(py),
652 (PyEnsureFuture {
653 awaitable: awaitable.into(),
654 tx: Some(tx),
655 },),
656 )?;
657
658 Ok(async move {
659 match rx.await {
660 Ok(item) => item,
661 Err(_) => Python::with_gil(|py| {
662 Err(PyErr::from_value(
663 asyncio(py)?.call_method0("CancelledError")?,
664 ))
665 }),
666 }
667 })
668}
669
670fn dump_err(py: Python<'_>) -> impl FnOnce(PyErr) + '_ {
671 move |e| {
672 // We can't display Python exceptions via std::fmt::Display,
673 // so print the error here manually.
674 e.print_and_set_sys_last_vars(py);
675 }
676}