Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/from_thread.py: 29%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3__all__ = (
4 "BlockingPortal",
5 "BlockingPortalProvider",
6 "check_cancelled",
7 "run",
8 "run_sync",
9 "start_blocking_portal",
10)
12import sys
13from collections.abc import Awaitable, Callable, Coroutine, Generator
14from concurrent.futures import Future
15from contextlib import (
16 AbstractAsyncContextManager,
17 AbstractContextManager,
18 contextmanager,
19)
20from contextvars import Context
21from dataclasses import dataclass, field
22from functools import partial
23from inspect import isawaitable
24from threading import Lock, Thread, current_thread, get_ident
25from types import TracebackType
26from typing import (
27 Any,
28 Generic,
29 TypeVar,
30 cast,
31 overload,
32)
34from ._core._eventloop import (
35 get_cancelled_exc_class,
36 threadlocals,
37)
38from ._core._eventloop import run as run_eventloop
39from ._core._exceptions import NoEventLoopError, RunFinishedError
40from ._core._synchronization import Event
41from ._core._tasks import CancelScope, create_task_group
42from .abc._tasks import TaskStatus
43from .lowlevel import EventLoopToken, current_token
45if sys.version_info >= (3, 11):
46 from typing import Self, TypeVarTuple, Unpack
47else:
48 from typing_extensions import Self, TypeVarTuple, Unpack
50T_Retval = TypeVar("T_Retval")
51T_co = TypeVar("T_co", covariant=True)
52PosArgsT = TypeVarTuple("PosArgsT")
55def _token_or_error(token: EventLoopToken | None) -> EventLoopToken:
56 if token is not None:
57 return token
59 try:
60 return threadlocals.current_token
61 except AttributeError:
62 raise NoEventLoopError(
63 "Not running inside an AnyIO worker thread, and no event loop token was "
64 "provided"
65 ) from None
68def run(
69 func: Callable[[Unpack[PosArgsT]], Coroutine[Any, Any, T_co]],
70 *args: Unpack[PosArgsT],
71 token: EventLoopToken | None = None,
72) -> T_co:
73 """
74 Call a coroutine function from a worker thread.
76 :param func: a coroutine function
77 :param args: positional arguments for the callable
78 :param token: an event loop token to use to get back to the event loop thread
79 (required if calling this function from outside an AnyIO worker thread)
80 :return: the return value of the coroutine function
81 :raises MissingTokenError: if no token was provided and called from outside an
82 AnyIO worker thread
83 :raises RunFinishedError: if the event loop tied to ``token`` is no longer running
85 .. versionchanged:: 4.11.0
86 Added the ``token`` parameter.
88 """
89 explicit_token = token is not None
90 token = _token_or_error(token)
91 return token.backend_class.run_async_from_thread(
92 func, args, token=token.native_token if explicit_token else None
93 )
96def run_sync(
97 func: Callable[[Unpack[PosArgsT]], T_Retval],
98 *args: Unpack[PosArgsT],
99 token: EventLoopToken | None = None,
100) -> T_Retval:
101 """
102 Call a function in the event loop thread from a worker thread.
104 :param func: a callable
105 :param args: positional arguments for the callable
106 :param token: an event loop token to use to get back to the event loop thread
107 (required if calling this function from outside an AnyIO worker thread)
108 :return: the return value of the callable
109 :raises MissingTokenError: if no token was provided and called from outside an
110 AnyIO worker thread
111 :raises RunFinishedError: if the event loop tied to ``token`` is no longer running
113 .. versionchanged:: 4.11.0
114 Added the ``token`` parameter.
116 """
117 explicit_token = token is not None
118 token = _token_or_error(token)
119 return token.backend_class.run_sync_from_thread(
120 func, args, token=token.native_token if explicit_token else None
121 )
124class _BlockingAsyncContextManager(AbstractContextManager, Generic[T_co]):
125 _enter_future: Future[T_co]
126 _exit_future: Future[bool | None]
127 _exit_event: Event
128 _exit_exc_info: tuple[
129 type[BaseException] | None, BaseException | None, TracebackType | None
130 ] = (None, None, None)
132 def __init__(
133 self, async_cm: AbstractAsyncContextManager[T_co], portal: BlockingPortal
134 ):
135 self._async_cm = async_cm
136 self._portal = portal
138 async def run_async_cm(self) -> bool | None:
139 try:
140 self._exit_event = Event()
141 value = await self._async_cm.__aenter__()
142 except BaseException as exc:
143 self._enter_future.set_exception(exc)
144 raise
145 else:
146 self._enter_future.set_result(value)
148 try:
149 # Wait for the sync context manager to exit.
150 # This next statement can raise `get_cancelled_exc_class()` if
151 # something went wrong in a task group in this async context
152 # manager.
153 await self._exit_event.wait()
154 finally:
155 # In case of cancellation, it could be that we end up here before
156 # `_BlockingAsyncContextManager.__exit__` is called, and an
157 # `_exit_exc_info` has been set.
158 result = await self._async_cm.__aexit__(*self._exit_exc_info)
160 return result
162 def __enter__(self) -> T_co:
163 self._enter_future = Future()
164 self._exit_future = self._portal.start_task_soon(self.run_async_cm)
165 return self._enter_future.result()
167 def __exit__(
168 self,
169 exc_type: type[BaseException] | None,
170 exc_value: BaseException | None,
171 traceback: TracebackType | None,
172 /,
173 ) -> bool | None:
174 self._exit_exc_info = exc_type, exc_value, traceback
175 self._portal.call(self._exit_event.set)
176 return self._exit_future.result()
179class _BlockingPortalTaskStatus(TaskStatus):
180 def __init__(self, future: Future):
181 self._future = future
183 def started(self, value: object = None) -> None:
184 self._future.set_result(value)
187class BlockingPortal:
188 """
189 An object that lets external threads run code in an asynchronous event loop.
191 :raises NoEventLoopError: if no supported asynchronous event loop is running in the
192 current thread
193 """
195 def __init__(self) -> None:
196 self._token = current_token()
197 self._event_loop_thread_id: int | None = get_ident()
198 self._stop_event = Event()
199 self._task_group = create_task_group()
201 async def __aenter__(self) -> Self:
202 await self._task_group.__aenter__()
203 return self
205 async def __aexit__(
206 self,
207 exc_type: type[BaseException] | None,
208 exc_val: BaseException | None,
209 exc_tb: TracebackType | None,
210 ) -> bool:
211 await self.stop()
212 return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
214 def _check_running(self) -> None:
215 if self._event_loop_thread_id is None:
216 raise RuntimeError("This portal is not running")
217 if self._event_loop_thread_id == get_ident():
218 raise RuntimeError(
219 "This method cannot be called from the event loop thread"
220 )
222 async def sleep_until_stopped(self) -> None:
223 """Sleep until :meth:`stop` is called."""
224 await self._stop_event.wait()
226 async def stop(self, cancel_remaining: bool = False) -> None:
227 """
228 Signal the portal to shut down.
230 This marks the portal as no longer accepting new calls and exits from
231 :meth:`sleep_until_stopped`.
233 :param cancel_remaining: ``True`` to cancel all the remaining tasks, ``False``
234 to let them finish before returning
236 """
237 self._event_loop_thread_id = None
238 self._stop_event.set()
239 if cancel_remaining:
240 self._task_group.cancel_scope.cancel("the blocking portal is shutting down")
242 async def _call_func(
243 self,
244 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
245 args: tuple[Unpack[PosArgsT]],
246 kwargs: dict[str, Any],
247 future: Future[T_Retval],
248 ) -> None:
249 event_loop_thread_id = self._event_loop_thread_id
251 def callback(f: Future[T_Retval]) -> None:
252 if f.cancelled():
253 if event_loop_thread_id == get_ident():
254 scope.cancel("the future was cancelled")
255 elif event_loop_thread_id is not None:
256 try:
257 self._token.backend_class.run_sync_soon(
258 scope.cancel,
259 ("the future was cancelled",),
260 self._token.native_token,
261 )
262 except RunFinishedError:
263 # The target event loop is gone, so there's nothing left to
264 # cancel or notify
265 pass
267 try:
268 retval_or_awaitable = func(*args, **kwargs)
269 if isawaitable(retval_or_awaitable):
270 with CancelScope() as scope:
271 future.add_done_callback(callback)
272 retval = await retval_or_awaitable
273 else:
274 retval = retval_or_awaitable
275 except get_cancelled_exc_class():
276 future.cancel()
277 future.set_running_or_notify_cancel()
278 except BaseException as exc:
279 if future.set_running_or_notify_cancel():
280 future.set_exception(exc)
282 # Let base exceptions fall through
283 if not isinstance(exc, Exception):
284 raise
285 else:
286 if future.set_running_or_notify_cancel():
287 future.set_result(retval)
288 finally:
289 scope = None # type: ignore[assignment]
291 def _spawn_task_from_thread(
292 self,
293 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
294 args: tuple[Unpack[PosArgsT]],
295 kwargs: dict[str, Any],
296 name: object,
297 future: Future[T_Retval],
298 ) -> None:
299 """
300 Spawn a new task using the given callable.
302 :param func: a callable
303 :param args: positional arguments to be passed to the callable
304 :param kwargs: keyword arguments to be passed to the callable
305 :param name: name of the task (will be coerced to a string if not ``None``)
306 :param future: a future that will resolve to the return value of the callable,
307 or the exception raised during its execution
309 """
310 run_sync(
311 partial(self._task_group.start_soon, name=name),
312 self._call_func,
313 func,
314 args,
315 kwargs,
316 future,
317 token=self._token,
318 )
320 @overload
321 def call(
322 self,
323 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
324 *args: Unpack[PosArgsT],
325 ) -> T_Retval: ...
327 @overload
328 def call(
329 self, func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT]
330 ) -> T_Retval: ...
332 def call(
333 self,
334 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
335 *args: Unpack[PosArgsT],
336 ) -> T_Retval:
337 """
338 Call the given function in the event loop thread.
340 If the callable returns a coroutine object, it is awaited on.
342 :param func: any callable
343 :raises RuntimeError: if the portal is not running or if this method is called
344 from within the event loop thread
346 """
347 return cast(T_Retval, self.start_task_soon(func, *args).result())
349 @overload
350 async def call_async(
351 self,
352 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
353 *args: Unpack[PosArgsT],
354 abandon_on_cancel: bool = False,
355 ) -> T_Retval: ...
357 @overload
358 async def call_async(
359 self,
360 func: Callable[[Unpack[PosArgsT]], T_Retval],
361 *args: Unpack[PosArgsT],
362 abandon_on_cancel: bool = False,
363 ) -> T_Retval: ...
365 async def call_async(
366 self,
367 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
368 *args: Unpack[PosArgsT],
369 abandon_on_cancel: bool = False,
370 ) -> T_Retval:
371 """
372 Call the given function in the portal's event loop from another event loop.
374 If the callable returns an awaitable, it is awaited on. Cancelling the calling
375 task cancels the task running in the portal.
377 :param func: any callable
378 :param abandon_on_cancel: if ``False``, wait for the cancelled task to finish
379 and propagate any exception raised during cleanup, or otherwise re-raise
380 the caller's cancellation; if ``True``, stop waiting immediately and ignore
381 the task's subsequent return value or exception
382 :raises RuntimeError: if the portal is not running or if this method is called
383 from within the portal's event loop thread
385 .. versionadded:: 4.16.0
386 """
387 self._check_running()
388 source_token = current_token()
389 event = Event()
390 future: Future[T_Retval] = Future()
391 cancellation: Future[None] = Future()
393 async def run_call() -> None:
394 with CancelScope() as scope:
396 def cancel_target(_: Future[None]) -> None:
397 try:
398 self._token.backend_class.run_sync_soon(
399 scope.cancel,
400 ("the calling task was cancelled",),
401 self._token.native_token,
402 )
403 except RunFinishedError:
404 pass
406 cancellation.add_done_callback(cancel_target)
407 await self._call_func(func, args, {}, future)
409 def notify_source(_: Future[T_Retval]) -> None:
410 try:
411 source_token.backend_class.run_sync_soon(
412 event.set, (), source_token.native_token
413 )
414 except RunFinishedError:
415 pass
417 def spawn_task() -> None:
418 try:
419 self._task_group.start_soon(run_call)
420 except BaseException as exc:
421 if not future.done():
422 future.set_exception(exc)
424 future.add_done_callback(notify_source)
425 self._token.backend_class.run_sync_soon(
426 spawn_task, (), self._token.native_token
427 )
428 try:
429 await event.wait()
430 except get_cancelled_exc_class():
431 cancellation.cancel()
432 if not abandon_on_cancel:
433 with CancelScope(shield=True):
434 await event.wait()
436 if not future.cancelled():
437 future.result()
439 raise
441 return future.result()
443 @overload
444 def start_task_soon(
445 self,
446 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
447 *args: Unpack[PosArgsT],
448 name: object = None,
449 ) -> Future[T_Retval]: ...
451 @overload
452 def start_task_soon(
453 self,
454 func: Callable[[Unpack[PosArgsT]], T_Retval],
455 *args: Unpack[PosArgsT],
456 name: object = None,
457 ) -> Future[T_Retval]: ...
459 def start_task_soon(
460 self,
461 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
462 *args: Unpack[PosArgsT],
463 name: object = None,
464 ) -> Future[T_Retval]:
465 """
466 Start a task in the portal's task group.
468 The task will be run inside a cancel scope which can be cancelled by cancelling
469 the returned future.
471 :param func: the target function
472 :param args: positional arguments passed to ``func``
473 :param name: name of the task (will be coerced to a string if not ``None``)
474 :return: a future that resolves with the return value of the callable if the
475 task completes successfully, or with the exception raised in the task
476 :raises RuntimeError: if the portal is not running or if this method is called
477 from within the event loop thread
478 :rtype: concurrent.futures.Future[T_Retval]
480 .. versionadded:: 3.0
482 """
483 self._check_running()
484 f: Future[T_Retval] = Future()
485 self._spawn_task_from_thread(func, args, {}, name, f)
486 return f
488 def start_task(
489 self,
490 func: Callable[..., Awaitable[T_Retval]],
491 *args: object,
492 name: object = None,
493 ) -> tuple[Future[T_Retval], Any]:
494 """
495 Start a task in the portal's task group and wait until it signals for readiness.
497 This method works the same way as :meth:`.abc.TaskGroup.start`.
499 :param func: the target function
500 :param args: positional arguments passed to ``func``
501 :param name: name of the task (will be coerced to a string if not ``None``)
502 :return: a tuple of (future, task_status_value) where the ``task_status_value``
503 is the value passed to ``task_status.started()`` from within the target
504 function
505 :rtype: tuple[concurrent.futures.Future[T_Retval], Any]
507 .. versionadded:: 3.0
509 """
511 def task_done(future: Future[T_Retval]) -> None:
512 if not task_status_future.done():
513 if future.cancelled():
514 task_status_future.cancel()
515 elif future.exception():
516 task_status_future.set_exception(future.exception())
517 else:
518 exc = RuntimeError(
519 "Task exited without calling task_status.started()"
520 )
521 task_status_future.set_exception(exc)
523 self._check_running()
524 task_status_future: Future = Future()
525 task_status = _BlockingPortalTaskStatus(task_status_future)
526 f: Future = Future()
527 f.add_done_callback(task_done)
528 self._spawn_task_from_thread(func, args, {"task_status": task_status}, name, f)
529 return f, task_status_future.result()
531 def wrap_async_context_manager(
532 self, cm: AbstractAsyncContextManager[T_co]
533 ) -> AbstractContextManager[T_co]:
534 """
535 Wrap an async context manager as a synchronous context manager via this portal.
537 Spawns a task that will call both ``__aenter__()`` and ``__aexit__()``, stopping
538 in the middle until the synchronous context manager exits.
540 :param cm: an asynchronous context manager
541 :return: a synchronous context manager
543 .. versionadded:: 2.1
545 """
546 return _BlockingAsyncContextManager(cm, self)
549@dataclass
550class BlockingPortalProvider:
551 """
552 A manager for a blocking portal. Used as a context manager. The first thread to
553 enter this context manager causes a blocking portal to be started with the specific
554 parameters, and the last thread to exit causes the portal to be shut down. Thus,
555 there will be exactly one blocking portal running in this context as long as at
556 least one thread has entered this context manager.
558 The parameters are the same as for :func:`~anyio.run`.
560 :param backend: name of the backend
561 :param backend_options: backend options
563 .. versionadded:: 4.4
564 """
566 backend: str = "asyncio"
567 backend_options: dict[str, Any] | None = None
568 _lock: Lock = field(init=False, default_factory=Lock)
569 _leases: int = field(init=False, default=0)
570 _portal: BlockingPortal = field(init=False)
571 _portal_cm: AbstractContextManager[BlockingPortal] | None = field(
572 init=False, default=None
573 )
575 def __enter__(self) -> BlockingPortal:
576 with self._lock:
577 if self._portal_cm is None:
578 portal_cm = start_blocking_portal(self.backend, self.backend_options)
579 portal = portal_cm.__enter__()
580 self._portal_cm = portal_cm
581 self._portal = portal
583 self._leases += 1
584 return self._portal
586 def __exit__(
587 self,
588 exc_type: type[BaseException] | None,
589 exc_val: BaseException | None,
590 exc_tb: TracebackType | None,
591 ) -> None:
592 portal_cm: AbstractContextManager[BlockingPortal] | None = None
593 with self._lock:
594 assert self._portal_cm
595 assert self._leases > 0
596 self._leases -= 1
597 if not self._leases:
598 portal_cm = self._portal_cm
599 self._portal_cm = None
600 del self._portal
602 if portal_cm:
603 portal_cm.__exit__(exc_type, exc_val, exc_tb)
606@contextmanager
607def start_blocking_portal(
608 backend: str = "asyncio",
609 backend_options: dict[str, Any] | None = None,
610 *,
611 name: str | None = None,
612) -> Generator[BlockingPortal, Any, None]:
613 """
614 Start a new event loop in a new thread and run a blocking portal in its main task.
616 The parameters are the same as for :func:`~anyio.run`.
618 :param backend: name of the backend
619 :param backend_options: backend options
620 :param name: name of the thread
621 :return: a context manager that yields a blocking portal
623 .. versionchanged:: 3.0
624 Usage as a context manager is now required.
626 """
628 async def run_portal() -> None:
629 async with BlockingPortal() as portal_:
630 if name is None:
631 current_thread().name = f"{backend}-portal-{id(portal_):x}"
633 future.set_result(portal_)
634 await portal_.sleep_until_stopped()
636 def run_blocking_portal() -> None:
637 if future.set_running_or_notify_cancel():
638 try:
639 run_eventloop(
640 run_portal, backend=backend, backend_options=backend_options
641 )
642 except BaseException as exc:
643 if not future.done():
644 future.set_exception(exc)
646 future: Future[BlockingPortal] = Future()
647 kwargs: dict[str, Any] = {}
648 if sys.version_info >= (3, 14):
649 kwargs["context"] = Context()
651 thread = Thread(target=run_blocking_portal, daemon=True, name=name, **kwargs)
652 thread.start()
653 try:
654 cancel_remaining_tasks = False
655 portal = future.result()
656 try:
657 yield portal
658 except BaseException:
659 cancel_remaining_tasks = True
660 raise
661 finally:
662 try:
663 portal.call(portal.stop, cancel_remaining_tasks)
664 except RuntimeError:
665 pass
666 finally:
667 thread.join()
670def check_cancelled() -> None:
671 """
672 Check if the cancel scope of the host task's running the current worker thread has
673 been cancelled.
675 If the host task's current cancel scope has indeed been cancelled, the
676 backend-specific cancellation exception will be raised.
678 :raises RuntimeError: if the current thread was not spawned by
679 :func:`.to_thread.run_sync`
681 """
682 try:
683 token: EventLoopToken = threadlocals.current_token
684 except AttributeError:
685 raise NoEventLoopError(
686 "This function can only be called inside an AnyIO worker thread"
687 ) from None
689 token.backend_class.check_cancelled()