20266013391

Committed 16 Dec 2025 11:15AM UTC coverage: 92.699% (-0.002%) from 92.701%

Build # 20266013391

Build Type

Pull #1049

github

Committed by

web-flow

Commit Message

Merge de99dc12d into bac48fc03

Pull Request Pull Request #1049: Replaced the NoCurrentAsyncBackend exception with NoEventLoopError

Run Details

20 of 20 new or added lines in 6 files covered. (100.0%)

42 existing lines in 4 files now uncovered.

5853 of 6314 relevant lines covered (92.7%)

10.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.47

/src/anyio/lowlevel.py

from __future__ import annotations

__all__ = (
    "EventLoopToken",
    "RunvarToken",
    "RunVar",
    "checkpoint",
    "checkpoint_if_cancelled",
    "cancel_shielded_checkpoint",
    "current_token",
)

import enum
from dataclasses import dataclass
from types import TracebackType
from typing import Any, Generic, Literal, TypeVar, final, overload
from weakref import WeakKeyDictionary

from ._core._eventloop import get_async_backend
from .abc import AsyncBackend

T = TypeVar("T")
D = TypeVar("D")


async def checkpoint() -> None:
    """
    Check for cancellation and allow the scheduler to switch to another task.

    Equivalent to (but more efficient than)::

        await checkpoint_if_cancelled()
        await cancel_shielded_checkpoint()


    .. versionadded:: 3.0

    """
    await get_async_backend().checkpoint()


async def checkpoint_if_cancelled() -> None:
    """
    Enter a checkpoint if the enclosing cancel scope has been cancelled.

    This does not allow the scheduler to switch to a different task.

    .. versionadded:: 3.0

    """
    await get_async_backend().checkpoint_if_cancelled()


async def cancel_shielded_checkpoint() -> None:
    """
    Allow the scheduler to switch to another task but without checking for cancellation.

    Equivalent to (but potentially more efficient than)::

        with CancelScope(shield=True):
            await checkpoint()


    .. versionadded:: 3.0

    """
    await get_async_backend().cancel_shielded_checkpoint()


@final
@dataclass(frozen=True, repr=False)
class EventLoopToken:
    """
    An opaque object that holds a reference to an event loop.

    .. versionadded:: 4.11.0
    """

    backend_class: type[AsyncBackend]
    native_token: object


def current_token() -> EventLoopToken:
    """
    Return a token object that can be used to call code in the current event loop from
    another thread.

    .. versionadded:: 4.11.0

    """
    backend_class = get_async_backend()
    raw_token = backend_class.current_token()
    return EventLoopToken(backend_class, raw_token)


_run_vars: WeakKeyDictionary[object, dict[RunVar[Any], Any]] = WeakKeyDictionary()


class _NoValueSet(enum.Enum):
    NO_VALUE_SET = enum.auto()


class RunvarToken(Generic[T]):
    __slots__ = "_var", "_value", "_redeemed"

    def __init__(self, var: RunVar[T], value: T | Literal[_NoValueSet.NO_VALUE_SET]):
        self._var = var
        self._value: T | Literal[_NoValueSet.NO_VALUE_SET] = value
        self._redeemed = False

    def __enter__(self) -> RunvarToken[T]:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self._var.reset(self)


class RunVar(Generic[T]):
    """
    Like a :class:`~contextvars.ContextVar`, except scoped to the running event loop.

    Can be used as a context manager, Just like :class:`~contextvars.ContextVar`, that
    will reset the variable to its previous value when the context block is exited.
    """

    __slots__ = "_name", "_default"

    NO_VALUE_SET: Literal[_NoValueSet.NO_VALUE_SET] = _NoValueSet.NO_VALUE_SET

    def __init__(
        self, name: str, default: T | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
    ):
        self._name = name
        self._default = default

    @property
    def _current_vars(self) -> dict[RunVar[T], T]:
        native_token = current_token().native_token
        try:
            return _run_vars[native_token]
        except KeyError:
            run_vars = _run_vars[native_token] = {}
            return run_vars

    @overload
    def get(self, default: D) -> T | D: ...

    @overload
    def get(self) -> T: ...

    def get(
        self, default: D | Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
    ) -> T | D:
        try:
            return self._current_vars[self]
        except KeyError:
            if default is not RunVar.NO_VALUE_SET:
                return default
            elif self._default is not RunVar.NO_VALUE_SET:
                return self._default

        raise LookupError(
            f'Run variable "{self._name}" has no value and no default set'
        )

    def set(self, value: T) -> RunvarToken[T]:
        current_vars = self._current_vars
        token = RunvarToken(self, current_vars.get(self, RunVar.NO_VALUE_SET))
        current_vars[self] = value
        return token

    def reset(self, token: RunvarToken[T]) -> None:
        if token._var is not self:
            raise ValueError("This token does not belong to this RunVar")

        if token._redeemed:
            raise ValueError("This token has already been used")

        if token._value is _NoValueSet.NO_VALUE_SET:
            try:
                del self._current_vars[self]
            except KeyError:
                pass
        else:
            self._current_vars[self] = token._value

        token._redeemed = True

    def __repr__(self) -> str:
        return f"<RunVar name={self._name!r}>"

1	from __future__ import annotations	12✔
2
3	__all__ = (	12✔
4	"EventLoopToken",
5	"RunvarToken",
6	"RunVar",
7	"checkpoint",
8	"checkpoint_if_cancelled",
9	"cancel_shielded_checkpoint",
10	"current_token",
11	)
12
13	import enum	12✔
14	from dataclasses import dataclass	12✔
15	from types import TracebackType	12✔
16	from typing import Any, Generic, Literal, TypeVar, final, overload	12✔
17	from weakref import WeakKeyDictionary	12✔
18
19	from ._core._eventloop import get_async_backend	12✔
20	from .abc import AsyncBackend	12✔
21
22	T = TypeVar("T")	12✔
23	D = TypeVar("D")	12✔
24
25
26	async def checkpoint() -> None:	12✔
27	"""
28	Check for cancellation and allow the scheduler to switch to another task.
29
30	Equivalent to (but more efficient than)::
31
32	await checkpoint_if_cancelled()
33	await cancel_shielded_checkpoint()
34
35
36	.. versionadded:: 3.0
37
38	"""
39	await get_async_backend().checkpoint()	12✔
40
41
42	async def checkpoint_if_cancelled() -> None:	12✔
43	"""
44	Enter a checkpoint if the enclosing cancel scope has been cancelled.
45
46	This does not allow the scheduler to switch to a different task.
47
48	.. versionadded:: 3.0
49
50	"""
51	await get_async_backend().checkpoint_if_cancelled()	12✔
52
53
54	async def cancel_shielded_checkpoint() -> None:	12✔
55	"""
56	Allow the scheduler to switch to another task but without checking for cancellation.
57
58	Equivalent to (but potentially more efficient than)::
59
60	with CancelScope(shield=True):
61	await checkpoint()
62
63
64	.. versionadded:: 3.0
65
66	"""
67	await get_async_backend().cancel_shielded_checkpoint()	12✔
68
69
70	@final	12✔
71	@dataclass(frozen=True, repr=False)	12✔
72	class EventLoopToken:	12✔
73	"""
74	An opaque object that holds a reference to an event loop.
75
76	.. versionadded:: 4.11.0
77	"""
78
79	backend_class: type[AsyncBackend]	12✔
80	native_token: object	12✔
81
82
83	def current_token() -> EventLoopToken:	12✔
84	"""
85	Return a token object that can be used to call code in the current event loop from
86	another thread.
87
88	.. versionadded:: 4.11.0
89
90	"""
91	backend_class = get_async_backend()	12✔
92	raw_token = backend_class.current_token()	12✔
93	return EventLoopToken(backend_class, raw_token)	12✔
94
95
96	_run_vars: WeakKeyDictionary[object, dict[RunVar[Any], Any]] = WeakKeyDictionary()	12✔
97
98
99	class _NoValueSet(enum.Enum):	12✔
100	NO_VALUE_SET = enum.auto()	12✔
101
102
103	class RunvarToken(Generic[T]):	12✔
104	__slots__ = "_var", "_value", "_redeemed"	12✔
105
106	def __init__(self, var: RunVar[T], value: T \| Literal[_NoValueSet.NO_VALUE_SET]):	12✔
107	self._var = var	12✔
108	self._value: T \| Literal[_NoValueSet.NO_VALUE_SET] = value	12✔
109	self._redeemed = False	12✔
110
111	def __enter__(self) -> RunvarToken[T]:	12✔
112	return self	12✔
113
114	def __exit__(	12✔
115	self,
116	exc_type: type[BaseException] \| None,
117	exc_val: BaseException \| None,
118	exc_tb: TracebackType \| None,
119	) -> None:
120	self._var.reset(self)	12✔
121
122
123	class RunVar(Generic[T]):	12✔
124	"""
125	Like a :class:`~contextvars.ContextVar`, except scoped to the running event loop.
126
127	Can be used as a context manager, Just like :class:`~contextvars.ContextVar`, that
128	will reset the variable to its previous value when the context block is exited.
129	"""
130
131	__slots__ = "_name", "_default"	12✔
132
133	NO_VALUE_SET: Literal[_NoValueSet.NO_VALUE_SET] = _NoValueSet.NO_VALUE_SET	12✔
134
135	def __init__(	12✔
136	self, name: str, default: T \| Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
137	):
138	self._name = name	12✔
139	self._default = default	12✔
140
141	@property	12✔
142	def _current_vars(self) -> dict[RunVar[T], T]:	12✔
143	native_token = current_token().native_token	12✔
144	try:	12✔
145	return _run_vars[native_token]	12✔
146	except KeyError:	12✔
147	run_vars = _run_vars[native_token] = {}	12✔
148	return run_vars	12✔
149
150	@overload	1✔
151	def get(self, default: D) -> T \| D: ...	1✔
152
153	@overload	1✔
154	def get(self) -> T: ...	1✔
155
156	def get(	12✔
157	self, default: D \| Literal[_NoValueSet.NO_VALUE_SET] = NO_VALUE_SET
158	) -> T \| D:
159	try:	12✔
160	return self._current_vars[self]	12✔
161	except KeyError:	12✔
162	if default is not RunVar.NO_VALUE_SET:	12✔
163	return default	12✔
164	elif self._default is not RunVar.NO_VALUE_SET:	12✔
165	return self._default	12✔
166
167	raise LookupError(	12✔
168	f'Run variable "{self._name}" has no value and no default set'
169	)
170
171	def set(self, value: T) -> RunvarToken[T]:	12✔
172	current_vars = self._current_vars	12✔
173	token = RunvarToken(self, current_vars.get(self, RunVar.NO_VALUE_SET))	12✔
174	current_vars[self] = value	12✔
175	return token	12✔
176
177	def reset(self, token: RunvarToken[T]) -> None:	12✔
178	if token._var is not self:	12✔
179	raise ValueError("This token does not belong to this RunVar")	12✔
180
181	if token._redeemed:	12✔
182	raise ValueError("This token has already been used")	12✔
183
184	if token._value is _NoValueSet.NO_VALUE_SET:	12✔
185	try:	12✔
186	del self._current_vars[self]	12✔
UNCOV 187	except KeyError:	×
188	pass	×
189	else:
190	self._current_vars[self] = token._value	12✔
191
192	token._redeemed = True	12✔
193
194	def __repr__(self) -> str:	12✔
UNCOV 195	return f"<RunVar name={self._name!r}>"	×

agronholm / anyio / 20266013391

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous