Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/from_thread.py: 33%
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
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 run_sync(
257 scope.cancel, "the future was cancelled", token=self._token
258 )
260 try:
261 retval_or_awaitable = func(*args, **kwargs)
262 if isawaitable(retval_or_awaitable):
263 with CancelScope() as scope:
264 future.add_done_callback(callback)
265 retval = await retval_or_awaitable
266 else:
267 retval = retval_or_awaitable
268 except get_cancelled_exc_class():
269 future.cancel()
270 future.set_running_or_notify_cancel()
271 except BaseException as exc:
272 if not future.cancelled():
273 future.set_exception(exc)
275 # Let base exceptions fall through
276 if not isinstance(exc, Exception):
277 raise
278 else:
279 if not future.cancelled():
280 future.set_result(retval)
281 finally:
282 scope = None # type: ignore[assignment]
284 def _spawn_task_from_thread(
285 self,
286 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
287 args: tuple[Unpack[PosArgsT]],
288 kwargs: dict[str, Any],
289 name: object,
290 future: Future[T_Retval],
291 ) -> None:
292 """
293 Spawn a new task using the given callable.
295 :param func: a callable
296 :param args: positional arguments to be passed to the callable
297 :param kwargs: keyword arguments to be passed to the callable
298 :param name: name of the task (will be coerced to a string if not ``None``)
299 :param future: a future that will resolve to the return value of the callable,
300 or the exception raised during its execution
302 """
303 run_sync(
304 partial(self._task_group.start_soon, name=name),
305 self._call_func,
306 func,
307 args,
308 kwargs,
309 future,
310 token=self._token,
311 )
313 @overload
314 def call(
315 self,
316 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
317 *args: Unpack[PosArgsT],
318 ) -> T_Retval: ...
320 @overload
321 def call(
322 self, func: Callable[[Unpack[PosArgsT]], T_Retval], *args: Unpack[PosArgsT]
323 ) -> T_Retval: ...
325 def call(
326 self,
327 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
328 *args: Unpack[PosArgsT],
329 ) -> T_Retval:
330 """
331 Call the given function in the event loop thread.
333 If the callable returns a coroutine object, it is awaited on.
335 :param func: any callable
336 :raises RuntimeError: if the portal is not running or if this method is called
337 from within the event loop thread
339 """
340 return cast(T_Retval, self.start_task_soon(func, *args).result())
342 @overload
343 def start_task_soon(
344 self,
345 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]],
346 *args: Unpack[PosArgsT],
347 name: object = None,
348 ) -> Future[T_Retval]: ...
350 @overload
351 def start_task_soon(
352 self,
353 func: Callable[[Unpack[PosArgsT]], T_Retval],
354 *args: Unpack[PosArgsT],
355 name: object = None,
356 ) -> Future[T_Retval]: ...
358 def start_task_soon(
359 self,
360 func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval] | T_Retval],
361 *args: Unpack[PosArgsT],
362 name: object = None,
363 ) -> Future[T_Retval]:
364 """
365 Start a task in the portal's task group.
367 The task will be run inside a cancel scope which can be cancelled by cancelling
368 the returned future.
370 :param func: the target function
371 :param args: positional arguments passed to ``func``
372 :param name: name of the task (will be coerced to a string if not ``None``)
373 :return: a future that resolves with the return value of the callable if the
374 task completes successfully, or with the exception raised in the task
375 :raises RuntimeError: if the portal is not running or if this method is called
376 from within the event loop thread
377 :rtype: concurrent.futures.Future[T_Retval]
379 .. versionadded:: 3.0
381 """
382 self._check_running()
383 f: Future[T_Retval] = Future()
384 self._spawn_task_from_thread(func, args, {}, name, f)
385 return f
387 def start_task(
388 self,
389 func: Callable[..., Awaitable[T_Retval]],
390 *args: object,
391 name: object = None,
392 ) -> tuple[Future[T_Retval], Any]:
393 """
394 Start a task in the portal's task group and wait until it signals for readiness.
396 This method works the same way as :meth:`.abc.TaskGroup.start`.
398 :param func: the target function
399 :param args: positional arguments passed to ``func``
400 :param name: name of the task (will be coerced to a string if not ``None``)
401 :return: a tuple of (future, task_status_value) where the ``task_status_value``
402 is the value passed to ``task_status.started()`` from within the target
403 function
404 :rtype: tuple[concurrent.futures.Future[T_Retval], Any]
406 .. versionadded:: 3.0
408 """
410 def task_done(future: Future[T_Retval]) -> None:
411 if not task_status_future.done():
412 if future.cancelled():
413 task_status_future.cancel()
414 elif future.exception():
415 task_status_future.set_exception(future.exception())
416 else:
417 exc = RuntimeError(
418 "Task exited without calling task_status.started()"
419 )
420 task_status_future.set_exception(exc)
422 self._check_running()
423 task_status_future: Future = Future()
424 task_status = _BlockingPortalTaskStatus(task_status_future)
425 f: Future = Future()
426 f.add_done_callback(task_done)
427 self._spawn_task_from_thread(func, args, {"task_status": task_status}, name, f)
428 return f, task_status_future.result()
430 def wrap_async_context_manager(
431 self, cm: AbstractAsyncContextManager[T_co]
432 ) -> AbstractContextManager[T_co]:
433 """
434 Wrap an async context manager as a synchronous context manager via this portal.
436 Spawns a task that will call both ``__aenter__()`` and ``__aexit__()``, stopping
437 in the middle until the synchronous context manager exits.
439 :param cm: an asynchronous context manager
440 :return: a synchronous context manager
442 .. versionadded:: 2.1
444 """
445 return _BlockingAsyncContextManager(cm, self)
448@dataclass
449class BlockingPortalProvider:
450 """
451 A manager for a blocking portal. Used as a context manager. The first thread to
452 enter this context manager causes a blocking portal to be started with the specific
453 parameters, and the last thread to exit causes the portal to be shut down. Thus,
454 there will be exactly one blocking portal running in this context as long as at
455 least one thread has entered this context manager.
457 The parameters are the same as for :func:`~anyio.run`.
459 :param backend: name of the backend
460 :param backend_options: backend options
462 .. versionadded:: 4.4
463 """
465 backend: str = "asyncio"
466 backend_options: dict[str, Any] | None = None
467 _lock: Lock = field(init=False, default_factory=Lock)
468 _leases: int = field(init=False, default=0)
469 _portal: BlockingPortal = field(init=False)
470 _portal_cm: AbstractContextManager[BlockingPortal] | None = field(
471 init=False, default=None
472 )
474 def __enter__(self) -> BlockingPortal:
475 with self._lock:
476 if self._portal_cm is None:
477 self._portal_cm = start_blocking_portal(
478 self.backend, self.backend_options
479 )
480 self._portal = self._portal_cm.__enter__()
482 self._leases += 1
483 return self._portal
485 def __exit__(
486 self,
487 exc_type: type[BaseException] | None,
488 exc_val: BaseException | None,
489 exc_tb: TracebackType | None,
490 ) -> None:
491 portal_cm: AbstractContextManager[BlockingPortal] | None = None
492 with self._lock:
493 assert self._portal_cm
494 assert self._leases > 0
495 self._leases -= 1
496 if not self._leases:
497 portal_cm = self._portal_cm
498 self._portal_cm = None
499 del self._portal
501 if portal_cm:
502 portal_cm.__exit__(None, None, None)
505@contextmanager
506def start_blocking_portal(
507 backend: str = "asyncio",
508 backend_options: dict[str, Any] | None = None,
509 *,
510 name: str | None = None,
511) -> Generator[BlockingPortal, Any, None]:
512 """
513 Start a new event loop in a new thread and run a blocking portal in its main task.
515 The parameters are the same as for :func:`~anyio.run`.
517 :param backend: name of the backend
518 :param backend_options: backend options
519 :param name: name of the thread
520 :return: a context manager that yields a blocking portal
522 .. versionchanged:: 3.0
523 Usage as a context manager is now required.
525 """
527 async def run_portal() -> None:
528 async with BlockingPortal() as portal_:
529 if name is None:
530 current_thread().name = f"{backend}-portal-{id(portal_):x}"
532 future.set_result(portal_)
533 await portal_.sleep_until_stopped()
535 def run_blocking_portal() -> None:
536 if future.set_running_or_notify_cancel():
537 try:
538 run_eventloop(
539 run_portal, backend=backend, backend_options=backend_options
540 )
541 except BaseException as exc:
542 if not future.done():
543 future.set_exception(exc)
545 future: Future[BlockingPortal] = Future()
546 kwargs: dict[str, Any] = {}
547 if sys.version_info >= (3, 14):
548 kwargs["context"] = Context()
550 thread = Thread(target=run_blocking_portal, daemon=True, name=name, **kwargs)
551 thread.start()
552 try:
553 cancel_remaining_tasks = False
554 portal = future.result()
555 try:
556 yield portal
557 except BaseException:
558 cancel_remaining_tasks = True
559 raise
560 finally:
561 try:
562 portal.call(portal.stop, cancel_remaining_tasks)
563 except RuntimeError:
564 pass
565 finally:
566 thread.join()
569def check_cancelled() -> None:
570 """
571 Check if the cancel scope of the host task's running the current worker thread has
572 been cancelled.
574 If the host task's current cancel scope has indeed been cancelled, the
575 backend-specific cancellation exception will be raised.
577 :raises RuntimeError: if the current thread was not spawned by
578 :func:`.to_thread.run_sync`
580 """
581 try:
582 token: EventLoopToken = threadlocals.current_token
583 except AttributeError:
584 raise NoEventLoopError(
585 "This function can only be called inside an AnyIO worker thread"
586 ) from None
588 token.backend_class.check_cancelled()