10708046317

Committed 04 Sep 2024 07:12PM UTC coverage: 91.676% (-0.2%) from 91.834%

Build # 10708046317

Build Type

Pull #761

github

Committed by

web-flow

Commit Message

Merge fa137ec93 into ee8165b55

Pull Request Pull Request #761: Delegated the implementations of Lock and Semaphore to the async backend class

Run Details

249 of 274 new or added lines in 4 files covered. (90.88%)

2 existing lines in 2 files now uncovered.

4791 of 5226 relevant lines covered (91.68%)

9.5 hits per line

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

83.23

/src/anyio/_core/_synchronization.py

from __future__ import annotations

import math
from collections import deque
from dataclasses import dataclass
from types import TracebackType

from sniffio import AsyncLibraryNotFoundError

from ..lowlevel import checkpoint
from ._eventloop import get_async_backend
from ._exceptions import BusyResourceError
from ._tasks import CancelScope
from ._testing import TaskInfo, get_current_task


@dataclass(frozen=True)
class EventStatistics:
    """
    :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Event.wait`
    """

    tasks_waiting: int


@dataclass(frozen=True)
class CapacityLimiterStatistics:
    """
    :ivar int borrowed_tokens: number of tokens currently borrowed by tasks
    :ivar float total_tokens: total number of available tokens
    :ivar tuple borrowers: tasks or other objects currently holding tokens borrowed from
        this limiter
    :ivar int tasks_waiting: number of tasks waiting on
        :meth:`~.CapacityLimiter.acquire` or
        :meth:`~.CapacityLimiter.acquire_on_behalf_of`
    """

    borrowed_tokens: int
    total_tokens: float
    borrowers: tuple[object, ...]
    tasks_waiting: int


@dataclass(frozen=True)
class LockStatistics:
    """
    :ivar bool locked: flag indicating if this lock is locked or not
    :ivar ~anyio.TaskInfo owner: task currently holding the lock (or ``None`` if the
        lock is not held by any task)
    :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Lock.acquire`
    """

    locked: bool
    owner: TaskInfo | None
    tasks_waiting: int


@dataclass(frozen=True)
class ConditionStatistics:
    """
    :ivar int tasks_waiting: number of tasks blocked on :meth:`~.Condition.wait`
    :ivar ~anyio.LockStatistics lock_statistics: statistics of the underlying
        :class:`~.Lock`
    """

    tasks_waiting: int
    lock_statistics: LockStatistics


@dataclass(frozen=True)
class SemaphoreStatistics:
    """
    :ivar int tasks_waiting: number of tasks waiting on :meth:`~.Semaphore.acquire`

    """

    tasks_waiting: int


class Event:
    def __new__(cls) -> Event:
        try:
            return get_async_backend().create_event()
        except AsyncLibraryNotFoundError:
            return EventAdapter()

    def set(self) -> None:
        """Set the flag, notifying all listeners."""
        raise NotImplementedError

    def is_set(self) -> bool:
        """Return ``True`` if the flag is set, ``False`` if not."""
        raise NotImplementedError

    async def wait(self) -> None:
        """
        Wait until the flag has been set.

        If the flag has already been set when this method is called, it returns
        immediately.

        """
        raise NotImplementedError

    def statistics(self) -> EventStatistics:
        """Return statistics about the current state of this event."""
        raise NotImplementedError


class EventAdapter(Event):
    _internal_event: Event | None = None

    def __new__(cls) -> EventAdapter:
        return object.__new__(cls)

    @property
    def _event(self) -> Event:
        if self._internal_event is None:
            self._internal_event = get_async_backend().create_event()

        return self._internal_event

    def set(self) -> None:
        self._event.set()

    def is_set(self) -> bool:
        return self._internal_event is not None and self._internal_event.is_set()

    async def wait(self) -> None:
        await self._event.wait()

    def statistics(self) -> EventStatistics:
        if self._internal_event is None:
            return EventStatistics(tasks_waiting=0)

        return self._internal_event.statistics()


class Lock:
    def __new__(cls, *, fast_acquire: bool = False) -> Lock:
        try:
            return get_async_backend().create_lock(fast_acquire=fast_acquire)
        except AsyncLibraryNotFoundError:
            return LockAdapter(fast_acquire=fast_acquire)

    async def __aenter__(self) -> None:
        await self.acquire()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self.release()

    async def acquire(self) -> None:
        """Acquire the lock."""
        raise NotImplementedError

    def acquire_nowait(self) -> None:
        """
        Acquire the lock, without blocking.

        :raises ~anyio.WouldBlock: if the operation would block

        """
        raise NotImplementedError

    def release(self) -> None:
        """Release the lock."""
        raise NotImplementedError

    def locked(self) -> bool:
        """Return True if the lock is currently held."""
        raise NotImplementedError

    def statistics(self) -> LockStatistics:
        """
        Return statistics about the current state of this lock.

        .. versionadded:: 3.0
        """
        raise NotImplementedError


class LockAdapter(Lock):
    _internal_lock: Lock | None = None

    def __new__(cls, *, fast_acquire: bool = False) -> LockAdapter:
        return object.__new__(cls)

    def __init__(self, *, fast_acquire: bool = False):
        self._fast_acquire = fast_acquire

    @property
    def _lock(self) -> Lock:
        if self._internal_lock is None:
            self._internal_lock = get_async_backend().create_lock(
                fast_acquire=self._fast_acquire
            )

        return self._internal_lock

    async def __aenter__(self) -> None:
        await self._lock.acquire()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if self._internal_lock is not None:
            self._internal_lock.release()

    async def acquire(self) -> None:
        """Acquire the lock."""
        await self._lock.acquire()

    def acquire_nowait(self) -> None:
        """
        Acquire the lock, without blocking.

        :raises ~anyio.WouldBlock: if the operation would block

        """
        self._lock.acquire_nowait()

    def release(self) -> None:
        """Release the lock."""
        self._lock.release()

    def locked(self) -> bool:
        """Return True if the lock is currently held."""
        return self._lock.locked()

    def statistics(self) -> LockStatistics:
        """
        Return statistics about the current state of this lock.

        .. versionadded:: 3.0

        """
        if self._internal_lock is None:
            return LockStatistics(False, None, 0)

        return self._internal_lock.statistics()


class Condition:
    _owner_task: TaskInfo | None = None

    def __init__(self, lock: Lock | None = None):
        self._lock = lock or Lock()
        self._waiters: deque[Event] = deque()

    async def __aenter__(self) -> None:
        await self.acquire()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self.release()

    def _check_acquired(self) -> None:
        if self._owner_task != get_current_task():
            raise RuntimeError("The current task is not holding the underlying lock")

    async def acquire(self) -> None:
        """Acquire the underlying lock."""
        await self._lock.acquire()
        self._owner_task = get_current_task()

    def acquire_nowait(self) -> None:
        """
        Acquire the underlying lock, without blocking.

        :raises ~anyio.WouldBlock: if the operation would block

        """
        self._lock.acquire_nowait()
        self._owner_task = get_current_task()

    def release(self) -> None:
        """Release the underlying lock."""
        self._lock.release()

    def locked(self) -> bool:
        """Return True if the lock is set."""
        return self._lock.locked()

    def notify(self, n: int = 1) -> None:
        """Notify exactly n listeners."""
        self._check_acquired()
        for _ in range(n):
            try:
                event = self._waiters.popleft()
            except IndexError:
                break

            event.set()

    def notify_all(self) -> None:
        """Notify all the listeners."""
        self._check_acquired()
        for event in self._waiters:
            event.set()

        self._waiters.clear()

    async def wait(self) -> None:
        """Wait for a notification."""
        await checkpoint()
        event = Event()
        self._waiters.append(event)
        self.release()
        try:
            await event.wait()
        except BaseException:
            if not event.is_set():
                self._waiters.remove(event)

            raise
        finally:
            with CancelScope(shield=True):
                await self.acquire()

    def statistics(self) -> ConditionStatistics:
        """
        Return statistics about the current state of this condition.

        .. versionadded:: 3.0
        """
        return ConditionStatistics(len(self._waiters), self._lock.statistics())


class Semaphore:
    def __new__(
        cls,
        initial_value: int,
        *,
        max_value: int | None = None,
        fast_acquire: bool = False,
    ) -> Semaphore:
        try:
            return get_async_backend().create_semaphore(
                initial_value, max_value=max_value, fast_acquire=fast_acquire
            )
        except AsyncLibraryNotFoundError:
            return SemaphoreAdapter(initial_value, max_value=max_value)

    def __init__(
        self,
        initial_value: int,
        *,
        max_value: int | None = None,
        fast_acquire: bool = False,
    ):
        if not isinstance(initial_value, int):
            raise TypeError("initial_value must be an integer")
        if initial_value < 0:
            raise ValueError("initial_value must be >= 0")
        if max_value is not None:
            if not isinstance(max_value, int):
                raise TypeError("max_value must be an integer or None")
            if max_value < initial_value:
                raise ValueError(
                    "max_value must be equal to or higher than initial_value"
                )

        self._fast_acquire = fast_acquire

    async def __aenter__(self) -> Semaphore:
        await self.acquire()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self.release()

    async def acquire(self) -> None:
        """Decrement the semaphore value, blocking if necessary."""
        raise NotImplementedError

    def acquire_nowait(self) -> None:
        """
        Acquire the underlying lock, without blocking.

        :raises ~anyio.WouldBlock: if the operation would block

        """
        raise NotImplementedError

    def release(self) -> None:
        """Increment the semaphore value."""
        raise NotImplementedError

    @property
    def value(self) -> int:
        """The current value of the semaphore."""
        raise NotImplementedError

    @property
    def max_value(self) -> int | None:
        """The maximum value of the semaphore."""
        raise NotImplementedError

    def statistics(self) -> SemaphoreStatistics:
        """
        Return statistics about the current state of this semaphore.

        .. versionadded:: 3.0
        """
        raise NotImplementedError


class SemaphoreAdapter(Semaphore):
    _internal_semaphore: Semaphore | None = None

    def __new__(
        cls,
        initial_value: int,
        *,
        max_value: int | None = None,
        fast_acquire: bool = False,
    ) -> SemaphoreAdapter:
        return object.__new__(cls)

    def __init__(
        self,
        initial_value: int,
        *,
        max_value: int | None = None,
        fast_acquire: bool = False,
    ) -> None:
        super().__init__(initial_value, max_value=max_value, fast_acquire=fast_acquire)
        self._initial_value = initial_value
        self._max_value = max_value

    @property
    def _semaphore(self) -> Semaphore:
        if self._internal_semaphore is None:
            self._internal_semaphore = get_async_backend().create_semaphore(
                self._initial_value, max_value=self._max_value
            )

        return self._internal_semaphore

    async def acquire(self) -> None:
        await self._semaphore.acquire()

    def acquire_nowait(self) -> None:
        self._semaphore.acquire_nowait()

    def release(self) -> None:
        self._semaphore.release()

    @property
    def value(self) -> int:
        if self._internal_semaphore is None:
            return self._initial_value

        return self._semaphore.value

    @property
    def max_value(self) -> int | None:
        return self._max_value

    def statistics(self) -> SemaphoreStatistics:
        if self._internal_semaphore is None:
            return SemaphoreStatistics(tasks_waiting=0)

        return self._semaphore.statistics()


class CapacityLimiter:
    def __new__(cls, total_tokens: float) -> CapacityLimiter:
        try:
            return get_async_backend().create_capacity_limiter(total_tokens)
        except AsyncLibraryNotFoundError:
            return CapacityLimiterAdapter(total_tokens)

    async def __aenter__(self) -> None:
        raise NotImplementedError

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool | None:
        raise NotImplementedError

    @property
    def total_tokens(self) -> float:
        """
        The total number of tokens available for borrowing.

        This is a read-write property. If the total number of tokens is increased, the
        proportionate number of tasks waiting on this limiter will be granted their
        tokens.

        .. versionchanged:: 3.0
            The property is now writable.

        """
        raise NotImplementedError

    @total_tokens.setter
    def total_tokens(self, value: float) -> None:
        raise NotImplementedError

    @property
    def borrowed_tokens(self) -> int:
        """The number of tokens that have currently been borrowed."""
        raise NotImplementedError

    @property
    def available_tokens(self) -> float:
        """The number of tokens currently available to be borrowed"""
        raise NotImplementedError

    def acquire_nowait(self) -> None:
        """
        Acquire a token for the current task without waiting for one to become
        available.

        :raises ~anyio.WouldBlock: if there are no tokens available for borrowing

        """
        raise NotImplementedError

    def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
        """
        Acquire a token without waiting for one to become available.

        :param borrower: the entity borrowing a token
        :raises ~anyio.WouldBlock: if there are no tokens available for borrowing

        """
        raise NotImplementedError

    async def acquire(self) -> None:
        """
        Acquire a token for the current task, waiting if necessary for one to become
        available.

        """
        raise NotImplementedError

    async def acquire_on_behalf_of(self, borrower: object) -> None:
        """
        Acquire a token, waiting if necessary for one to become available.

        :param borrower: the entity borrowing a token

        """
        raise NotImplementedError

    def release(self) -> None:
        """
        Release the token held by the current task.

        :raises RuntimeError: if the current task has not borrowed a token from this
            limiter.

        """
        raise NotImplementedError

    def release_on_behalf_of(self, borrower: object) -> None:
        """
        Release the token held by the given borrower.

        :raises RuntimeError: if the borrower has not borrowed a token from this
            limiter.

        """
        raise NotImplementedError

    def statistics(self) -> CapacityLimiterStatistics:
        """
        Return statistics about the current state of this limiter.

        .. versionadded:: 3.0

        """
        raise NotImplementedError


class CapacityLimiterAdapter(CapacityLimiter):
    _internal_limiter: CapacityLimiter | None = None

    def __new__(cls, total_tokens: float) -> CapacityLimiterAdapter:
        return object.__new__(cls)

    def __init__(self, total_tokens: float) -> None:
        self.total_tokens = total_tokens

    @property
    def _limiter(self) -> CapacityLimiter:
        if self._internal_limiter is None:
            self._internal_limiter = get_async_backend().create_capacity_limiter(
                self._total_tokens
            )

        return self._internal_limiter

    async def __aenter__(self) -> None:
        await self._limiter.__aenter__()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool | None:
        return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)

    @property
    def total_tokens(self) -> float:
        if self._internal_limiter is None:
            return self._total_tokens

        return self._internal_limiter.total_tokens

    @total_tokens.setter
    def total_tokens(self, value: float) -> None:
        if not isinstance(value, int) and value is not math.inf:
            raise TypeError("total_tokens must be an int or math.inf")
        elif value < 1:
            raise ValueError("total_tokens must be >= 1")

        if self._internal_limiter is None:
            self._total_tokens = value
            return

        self._limiter.total_tokens = value

    @property
    def borrowed_tokens(self) -> int:
        if self._internal_limiter is None:
            return 0

        return self._internal_limiter.borrowed_tokens

    @property
    def available_tokens(self) -> float:
        if self._internal_limiter is None:
            return self._total_tokens

        return self._internal_limiter.available_tokens

    def acquire_nowait(self) -> None:
        self._limiter.acquire_nowait()

    def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
        self._limiter.acquire_on_behalf_of_nowait(borrower)

    async def acquire(self) -> None:
        await self._limiter.acquire()

    async def acquire_on_behalf_of(self, borrower: object) -> None:
        await self._limiter.acquire_on_behalf_of(borrower)

    def release(self) -> None:
        self._limiter.release()

    def release_on_behalf_of(self, borrower: object) -> None:
        self._limiter.release_on_behalf_of(borrower)

    def statistics(self) -> CapacityLimiterStatistics:
        if self._internal_limiter is None:
            return CapacityLimiterStatistics(
                borrowed_tokens=0,
                total_tokens=self.total_tokens,
                borrowers=(),
                tasks_waiting=0,
            )

        return self._internal_limiter.statistics()


class ResourceGuard:
    """
    A context manager for ensuring that a resource is only used by a single task at a
    time.

    Entering this context manager while the previous has not exited it yet will trigger
    :exc:`BusyResourceError`.

    :param action: the action to guard against (visible in the :exc:`BusyResourceError`
        when triggered, e.g. "Another task is already {action} this resource")

    .. versionadded:: 4.1
    """

    __slots__ = "action", "_guarded"

    def __init__(self, action: str = "using"):
        self.action: str = action
        self._guarded = False

    def __enter__(self) -> None:
        if self._guarded:
            raise BusyResourceError(self.action)

        self._guarded = True

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool | None:
        self._guarded = False
        return None

1	from __future__ import annotations	11✔
2
3	import math	11✔
4	from collections import deque	11✔
5	from dataclasses import dataclass	11✔
6	from types import TracebackType	11✔
7
8	from sniffio import AsyncLibraryNotFoundError	11✔
9
10	from ..lowlevel import checkpoint	11✔
11	from ._eventloop import get_async_backend	11✔
12	from ._exceptions import BusyResourceError	11✔
13	from ._tasks import CancelScope	11✔
14	from ._testing import TaskInfo, get_current_task	11✔
15
16
17	@dataclass(frozen=True)	11✔
18	class EventStatistics:	11✔
19	"""
20	:ivar int tasks_waiting: number of tasks waiting on :meth:`~.Event.wait`
21	"""
22
23	tasks_waiting: int	11✔
24
25
26	@dataclass(frozen=True)	11✔
27	class CapacityLimiterStatistics:	11✔
28	"""
29	:ivar int borrowed_tokens: number of tokens currently borrowed by tasks
30	:ivar float total_tokens: total number of available tokens
31	:ivar tuple borrowers: tasks or other objects currently holding tokens borrowed from
32	this limiter
33	:ivar int tasks_waiting: number of tasks waiting on
34	:meth:`~.CapacityLimiter.acquire` or
35	:meth:`~.CapacityLimiter.acquire_on_behalf_of`
36	"""
37
38	borrowed_tokens: int	11✔
39	total_tokens: float	11✔
40	borrowers: tuple[object, ...]	11✔
41	tasks_waiting: int	11✔
42
43
44	@dataclass(frozen=True)	11✔
45	class LockStatistics:	11✔
46	"""
47	:ivar bool locked: flag indicating if this lock is locked or not
48	:ivar ~anyio.TaskInfo owner: task currently holding the lock (or ``None`` if the
49	lock is not held by any task)
50	:ivar int tasks_waiting: number of tasks waiting on :meth:`~.Lock.acquire`
51	"""
52
53	locked: bool	11✔
54	owner: TaskInfo \| None	11✔
55	tasks_waiting: int	11✔
56
57
58	@dataclass(frozen=True)	11✔
59	class ConditionStatistics:	11✔
60	"""
61	:ivar int tasks_waiting: number of tasks blocked on :meth:`~.Condition.wait`
62	:ivar ~anyio.LockStatistics lock_statistics: statistics of the underlying
63	:class:`~.Lock`
64	"""
65
66	tasks_waiting: int	11✔
67	lock_statistics: LockStatistics	11✔
68
69
70	@dataclass(frozen=True)	11✔
71	class SemaphoreStatistics:	11✔
72	"""
73	:ivar int tasks_waiting: number of tasks waiting on :meth:`~.Semaphore.acquire`
74
75	"""
76
77	tasks_waiting: int	11✔
78
79
80	class Event:	11✔
81	def __new__(cls) -> Event:	11✔
82	try:	11✔
83	return get_async_backend().create_event()	11✔
84	except AsyncLibraryNotFoundError:	10✔
85	return EventAdapter()	10✔
86
87	def set(self) -> None:	11✔
88	"""Set the flag, notifying all listeners."""
89	raise NotImplementedError	×
90
91	def is_set(self) -> bool:	11✔
92	"""Return ``True`` if the flag is set, ``False`` if not."""
93	raise NotImplementedError	×
94
95	async def wait(self) -> None:	11✔
96	"""
97	Wait until the flag has been set.
98
99	If the flag has already been set when this method is called, it returns
100	immediately.
101
102	"""
103	raise NotImplementedError	×
104
105	def statistics(self) -> EventStatistics:	11✔
106	"""Return statistics about the current state of this event."""
107	raise NotImplementedError	×
108
109
110	class EventAdapter(Event):	11✔
111	_internal_event: Event \| None = None	11✔
112
113	def __new__(cls) -> EventAdapter:	11✔
114	return object.__new__(cls)	10✔
115
116	@property	11✔
117	def _event(self) -> Event:	11✔
118	if self._internal_event is None:	11✔
119	self._internal_event = get_async_backend().create_event()	11✔
120
121	return self._internal_event	11✔
122
123	def set(self) -> None:	11✔
124	self._event.set()	11✔
125
126	def is_set(self) -> bool:	11✔
127	return self._internal_event is not None and self._internal_event.is_set()	10✔
128
129	async def wait(self) -> None:	11✔
130	await self._event.wait()	11✔
131
132	def statistics(self) -> EventStatistics:	11✔
133	if self._internal_event is None:	10✔
134	return EventStatistics(tasks_waiting=0)	10✔
135
136	return self._internal_event.statistics()	×
137
138
139	class Lock:	11✔
140	def __new__(cls, *, fast_acquire: bool = False) -> Lock:	11✔
141	try:	10✔
142	return get_async_backend().create_lock(fast_acquire=fast_acquire)	10✔
143	except AsyncLibraryNotFoundError:	10✔
144	return LockAdapter(fast_acquire=fast_acquire)	10✔
145
146	async def __aenter__(self) -> None:	11✔
147	await self.acquire()	10✔
148
149	async def __aexit__(	11✔
150	self,
151	exc_type: type[BaseException] \| None,
152	exc_val: BaseException \| None,
153	exc_tb: TracebackType \| None,
154	) -> None:
155	self.release()	10✔
156
157	async def acquire(self) -> None:	11✔
158	"""Acquire the lock."""
NEW 159	raise NotImplementedError	×
160
161	def acquire_nowait(self) -> None:	11✔
162	"""
163	Acquire the lock, without blocking.
164
165	:raises ~anyio.WouldBlock: if the operation would block
166
167	"""
NEW 168	raise NotImplementedError	×
169
170	def release(self) -> None:	11✔
171	"""Release the lock."""
NEW 172	raise NotImplementedError	×
173
174	def locked(self) -> bool:	11✔
175	"""Return True if the lock is currently held."""
NEW 176	raise NotImplementedError	×
177
178	def statistics(self) -> LockStatistics:	11✔
179	"""
180	Return statistics about the current state of this lock.
181
182	.. versionadded:: 3.0
183	"""
NEW 184	raise NotImplementedError	×
185
186
187	class LockAdapter(Lock):	11✔
188	_internal_lock: Lock \| None = None	11✔
189
190	def __new__(cls, *, fast_acquire: bool = False) -> LockAdapter:	11✔
191	return object.__new__(cls)	10✔
192
193	def __init__(self, *, fast_acquire: bool = False):	11✔
194	self._fast_acquire = fast_acquire	10✔
195
196	@property	11✔
197	def _lock(self) -> Lock:	11✔
198	if self._internal_lock is None:	10✔
199	self._internal_lock = get_async_backend().create_lock(	10✔
200	fast_acquire=self._fast_acquire
201	)
202
203	return self._internal_lock	10✔
204
205	async def __aenter__(self) -> None:	11✔
206	await self._lock.acquire()	10✔
207
208	async def __aexit__(	11✔
209	self,
210	exc_type: type[BaseException] \| None,
211	exc_val: BaseException \| None,
212	exc_tb: TracebackType \| None,
213	) -> None:
214	if self._internal_lock is not None:	10✔
215	self._internal_lock.release()	10✔
216
217	async def acquire(self) -> None:	11✔
218	"""Acquire the lock."""
219	await self._lock.acquire()	10✔
220
221	def acquire_nowait(self) -> None:	11✔
222	"""
223	Acquire the lock, without blocking.
224
225	:raises ~anyio.WouldBlock: if the operation would block
226
227	"""
NEW 228	self._lock.acquire_nowait()	×
229
230	def release(self) -> None:	11✔
231	"""Release the lock."""
232	self._lock.release()	10✔
233
234	def locked(self) -> bool:	11✔
235	"""Return True if the lock is currently held."""
NEW 236	return self._lock.locked()	×
237
238	def statistics(self) -> LockStatistics:	11✔
239	"""
240	Return statistics about the current state of this lock.
241
242	.. versionadded:: 3.0
243
244	"""
245	if self._internal_lock is None:	10✔
246	return LockStatistics(False, None, 0)	10✔
247
NEW 248	return self._internal_lock.statistics()	×
249
250
251	class Condition:	11✔
252	_owner_task: TaskInfo \| None = None	11✔
253
254	def __init__(self, lock: Lock \| None = None):	11✔
255	self._lock = lock or Lock()	10✔
256	self._waiters: deque[Event] = deque()	10✔
257
258	async def __aenter__(self) -> None:	11✔
259	await self.acquire()	10✔
260
261	async def __aexit__(	11✔
262	self,
263	exc_type: type[BaseException] \| None,
264	exc_val: BaseException \| None,
265	exc_tb: TracebackType \| None,
266	) -> None:
267	self.release()	10✔
268
269	def _check_acquired(self) -> None:	11✔
270	if self._owner_task != get_current_task():	10✔
271	raise RuntimeError("The current task is not holding the underlying lock")	×
272
273	async def acquire(self) -> None:	11✔
274	"""Acquire the underlying lock."""
275	await self._lock.acquire()	10✔
276	self._owner_task = get_current_task()	10✔
277
278	def acquire_nowait(self) -> None:	11✔
279	"""
280	Acquire the underlying lock, without blocking.
281
282	:raises ~anyio.WouldBlock: if the operation would block
283
284	"""
285	self._lock.acquire_nowait()	10✔
286	self._owner_task = get_current_task()	10✔
287
288	def release(self) -> None:	11✔
289	"""Release the underlying lock."""
290	self._lock.release()	10✔
291
292	def locked(self) -> bool:	11✔
293	"""Return True if the lock is set."""
294	return self._lock.locked()	10✔
295
296	def notify(self, n: int = 1) -> None:	11✔
297	"""Notify exactly n listeners."""
298	self._check_acquired()	10✔
299	for _ in range(n):	10✔
300	try:	10✔
301	event = self._waiters.popleft()	10✔
302	except IndexError:	×
303	break	×
304
305	event.set()	10✔
306
307	def notify_all(self) -> None:	11✔
308	"""Notify all the listeners."""
309	self._check_acquired()	10✔
310	for event in self._waiters:	10✔
311	event.set()	10✔
312
313	self._waiters.clear()	10✔
314
315	async def wait(self) -> None:	11✔
316	"""Wait for a notification."""
317	await checkpoint()	10✔
318	event = Event()	10✔
319	self._waiters.append(event)	10✔
320	self.release()	10✔
321	try:	10✔
322	await event.wait()	10✔
323	except BaseException:	10✔
324	if not event.is_set():	10✔
325	self._waiters.remove(event)	10✔
326
327	raise	10✔
328	finally:
329	with CancelScope(shield=True):	10✔
330	await self.acquire()	10✔
331
332	def statistics(self) -> ConditionStatistics:	11✔
333	"""
334	Return statistics about the current state of this condition.
335
336	.. versionadded:: 3.0
337	"""
338	return ConditionStatistics(len(self._waiters), self._lock.statistics())	10✔
339
340
341	class Semaphore:	11✔
342	def __new__(	11✔
343	cls,
344	initial_value: int,
345	*,
346	max_value: int \| None = None,
347	fast_acquire: bool = False,
348	) -> Semaphore:
349	try:	10✔
350	return get_async_backend().create_semaphore(	10✔
351	initial_value, max_value=max_value, fast_acquire=fast_acquire
352	)
353	except AsyncLibraryNotFoundError:	10✔
354	return SemaphoreAdapter(initial_value, max_value=max_value)	10✔
355
356	def __init__(	11✔
357	self,
358	initial_value: int,
359	*,
360	max_value: int \| None = None,
361	fast_acquire: bool = False,
362	):
363	if not isinstance(initial_value, int):	10✔
364	raise TypeError("initial_value must be an integer")	×
365	if initial_value < 0:	10✔
366	raise ValueError("initial_value must be >= 0")	×
367	if max_value is not None:	10✔
368	if not isinstance(max_value, int):	10✔
369	raise TypeError("max_value must be an integer or None")	×
370	if max_value < initial_value:	10✔
371	raise ValueError(	×
372	"max_value must be equal to or higher than initial_value"
373	)
374
375	self._fast_acquire = fast_acquire	10✔
376
377	async def __aenter__(self) -> Semaphore:	11✔
378	await self.acquire()	10✔
379	return self	10✔
380
381	async def __aexit__(	11✔
382	self,
383	exc_type: type[BaseException] \| None,
384	exc_val: BaseException \| None,
385	exc_tb: TracebackType \| None,
386	) -> None:
387	self.release()	10✔
388
389	async def acquire(self) -> None:	11✔
390	"""Decrement the semaphore value, blocking if necessary."""
NEW 391	raise NotImplementedError	×
392
393	def acquire_nowait(self) -> None:	11✔
394	"""
395	Acquire the underlying lock, without blocking.
396
397	:raises ~anyio.WouldBlock: if the operation would block
398
399	"""
NEW 400	raise NotImplementedError	×
401
402	def release(self) -> None:	11✔
403	"""Increment the semaphore value."""
NEW 404	raise NotImplementedError	×
405
406	@property	11✔
407	def value(self) -> int:	11✔
408	"""The current value of the semaphore."""
NEW 409	raise NotImplementedError	×
410
411	@property	11✔
412	def max_value(self) -> int \| None:	11✔
413	"""The maximum value of the semaphore."""
NEW 414	raise NotImplementedError	×
415
416	def statistics(self) -> SemaphoreStatistics:	11✔
417	"""
418	Return statistics about the current state of this semaphore.
419
420	.. versionadded:: 3.0
421	"""
NEW 422	raise NotImplementedError	×
423
424
425	class SemaphoreAdapter(Semaphore):	11✔
426	_internal_semaphore: Semaphore \| None = None	11✔
427
428	def __new__(	11✔
429	cls,
430	initial_value: int,
431	*,
432	max_value: int \| None = None,
433	fast_acquire: bool = False,
434	) -> SemaphoreAdapter:
435	return object.__new__(cls)	10✔
436
437	def __init__(	11✔
438	self,
439	initial_value: int,
440	*,
441	max_value: int \| None = None,
442	fast_acquire: bool = False,
443	) -> None:
444	super().__init__(initial_value, max_value=max_value, fast_acquire=fast_acquire)	10✔
445	self._initial_value = initial_value	10✔
446	self._max_value = max_value	10✔
447
448	@property	11✔
449	def _semaphore(self) -> Semaphore:	11✔
450	if self._internal_semaphore is None:	10✔
451	self._internal_semaphore = get_async_backend().create_semaphore(	10✔
452	self._initial_value, max_value=self._max_value
453	)
454
455	return self._internal_semaphore	10✔
456
457	async def acquire(self) -> None:	11✔
458	await self._semaphore.acquire()	10✔
459
460	def acquire_nowait(self) -> None:	11✔
NEW 461	self._semaphore.acquire_nowait()	×
462
463	def release(self) -> None:	11✔
464	self._semaphore.release()	10✔
465
466	@property	11✔
467	def value(self) -> int:	11✔
468	if self._internal_semaphore is None:	10✔
469	return self._initial_value	10✔
470
NEW 471	return self._semaphore.value	×
472
473	@property	11✔
474	def max_value(self) -> int \| None:	11✔
475	return self._max_value	10✔
476
477	def statistics(self) -> SemaphoreStatistics:	11✔
478	if self._internal_semaphore is None:	10✔
479	return SemaphoreStatistics(tasks_waiting=0)	10✔
480
NEW 481	return self._semaphore.statistics()	×
482
483
484	class CapacityLimiter:	11✔
485	def __new__(cls, total_tokens: float) -> CapacityLimiter:	11✔
486	try:	10✔
487	return get_async_backend().create_capacity_limiter(total_tokens)	10✔
488	except AsyncLibraryNotFoundError:	10✔
489	return CapacityLimiterAdapter(total_tokens)	10✔
490
491	async def __aenter__(self) -> None:	11✔
492	raise NotImplementedError	×
493
494	async def __aexit__(	11✔
495	self,
496	exc_type: type[BaseException] \| None,
497	exc_val: BaseException \| None,
498	exc_tb: TracebackType \| None,
499	) -> bool \| None:
500	raise NotImplementedError	×
501
502	@property	11✔
503	def total_tokens(self) -> float:	11✔
504	"""
505	The total number of tokens available for borrowing.
506
507	This is a read-write property. If the total number of tokens is increased, the
508	proportionate number of tasks waiting on this limiter will be granted their
509	tokens.
510
511	.. versionchanged:: 3.0
512	The property is now writable.
513
514	"""
515	raise NotImplementedError	×
516
517	@total_tokens.setter	11✔
518	def total_tokens(self, value: float) -> None:	11✔
519	raise NotImplementedError	×
520
521	@property	11✔
522	def borrowed_tokens(self) -> int:	11✔
523	"""The number of tokens that have currently been borrowed."""
524	raise NotImplementedError	×
525
526	@property	11✔
527	def available_tokens(self) -> float:	11✔
528	"""The number of tokens currently available to be borrowed"""
529	raise NotImplementedError	×
530
531	def acquire_nowait(self) -> None:	11✔
532	"""
533	Acquire a token for the current task without waiting for one to become
534	available.
535
536	:raises ~anyio.WouldBlock: if there are no tokens available for borrowing
537
538	"""
539	raise NotImplementedError	×
540
541	def acquire_on_behalf_of_nowait(self, borrower: object) -> None:	11✔
542	"""
543	Acquire a token without waiting for one to become available.
544
545	:param borrower: the entity borrowing a token
546	:raises ~anyio.WouldBlock: if there are no tokens available for borrowing
547
548	"""
549	raise NotImplementedError	×
550
551	async def acquire(self) -> None:	11✔
552	"""
553	Acquire a token for the current task, waiting if necessary for one to become
554	available.
555
556	"""
557	raise NotImplementedError	×
558
559	async def acquire_on_behalf_of(self, borrower: object) -> None:	11✔
560	"""
561	Acquire a token, waiting if necessary for one to become available.
562
563	:param borrower: the entity borrowing a token
564
565	"""
566	raise NotImplementedError	×
567
568	def release(self) -> None:	11✔
569	"""
570	Release the token held by the current task.
571
572	:raises RuntimeError: if the current task has not borrowed a token from this
573	limiter.
574
575	"""
576	raise NotImplementedError	×
577
578	def release_on_behalf_of(self, borrower: object) -> None:	11✔
579	"""
580	Release the token held by the given borrower.
581
582	:raises RuntimeError: if the borrower has not borrowed a token from this
583	limiter.
584
585	"""
586	raise NotImplementedError	×
587
588	def statistics(self) -> CapacityLimiterStatistics:	11✔
589	"""
590	Return statistics about the current state of this limiter.
591
592	.. versionadded:: 3.0
593
594	"""
595	raise NotImplementedError	×
596
597
598	class CapacityLimiterAdapter(CapacityLimiter):	11✔
599	_internal_limiter: CapacityLimiter \| None = None	11✔
600
601	def __new__(cls, total_tokens: float) -> CapacityLimiterAdapter:	11✔
602	return object.__new__(cls)	10✔
603
604	def __init__(self, total_tokens: float) -> None:	11✔
605	self.total_tokens = total_tokens	10✔
606
607	@property	11✔
608	def _limiter(self) -> CapacityLimiter:	11✔
609	if self._internal_limiter is None:	10✔
610	self._internal_limiter = get_async_backend().create_capacity_limiter(	10✔
611	self._total_tokens
612	)
613
614	return self._internal_limiter	10✔
615
616	async def __aenter__(self) -> None:	11✔
617	await self._limiter.__aenter__()	10✔
618
619	async def __aexit__(	11✔
620	self,
621	exc_type: type[BaseException] \| None,
622	exc_val: BaseException \| None,
623	exc_tb: TracebackType \| None,
624	) -> bool \| None:
625	return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)	10✔
626
627	@property	11✔
628	def total_tokens(self) -> float:	11✔
629	if self._internal_limiter is None:	10✔
630	return self._total_tokens	10✔
631
632	return self._internal_limiter.total_tokens	×
633
634	@total_tokens.setter	11✔
635	def total_tokens(self, value: float) -> None:	11✔
636	if not isinstance(value, int) and value is not math.inf:	10✔
637	raise TypeError("total_tokens must be an int or math.inf")	10✔
638	elif value < 1:	10✔
639	raise ValueError("total_tokens must be >= 1")	×
640
641	if self._internal_limiter is None:	10✔
642	self._total_tokens = value	10✔
643	return	10✔
644
645	self._limiter.total_tokens = value	×
646
647	@property	11✔
648	def borrowed_tokens(self) -> int:	11✔
649	if self._internal_limiter is None:	10✔
650	return 0	10✔
651
652	return self._internal_limiter.borrowed_tokens	×
653
654	@property	11✔
655	def available_tokens(self) -> float:	11✔
656	if self._internal_limiter is None:	×
657	return self._total_tokens	×
658
659	return self._internal_limiter.available_tokens	×
660
661	def acquire_nowait(self) -> None:	11✔
662	self._limiter.acquire_nowait()	×
663
664	def acquire_on_behalf_of_nowait(self, borrower: object) -> None:	11✔
665	self._limiter.acquire_on_behalf_of_nowait(borrower)	×
666
667	async def acquire(self) -> None:	11✔
668	await self._limiter.acquire()	×
669
670	async def acquire_on_behalf_of(self, borrower: object) -> None:	11✔
671	await self._limiter.acquire_on_behalf_of(borrower)	×
672
673	def release(self) -> None:	11✔
674	self._limiter.release()	×
675
676	def release_on_behalf_of(self, borrower: object) -> None:	11✔
677	self._limiter.release_on_behalf_of(borrower)	×
678
679	def statistics(self) -> CapacityLimiterStatistics:	11✔
680	if self._internal_limiter is None:	10✔
681	return CapacityLimiterStatistics(	10✔
682	borrowed_tokens=0,
683	total_tokens=self.total_tokens,
684	borrowers=(),
685	tasks_waiting=0,
686	)
687
688	return self._internal_limiter.statistics()	×
689
690
691	class ResourceGuard:	11✔
692	"""
693	A context manager for ensuring that a resource is only used by a single task at a
694	time.
695
696	Entering this context manager while the previous has not exited it yet will trigger
697	:exc:`BusyResourceError`.
698
699	:param action: the action to guard against (visible in the :exc:`BusyResourceError`
700	when triggered, e.g. "Another task is already {action} this resource")
701
702	.. versionadded:: 4.1
703	"""
704
705	__slots__ = "action", "_guarded"	11✔
706
707	def __init__(self, action: str = "using"):	11✔
708	self.action: str = action	11✔
709	self._guarded = False	11✔
710
711	def __enter__(self) -> None:	11✔
712	if self._guarded:	11✔
713	raise BusyResourceError(self.action)	11✔
714
715	self._guarded = True	11✔
716
717	def __exit__(	11✔
718	self,
719	exc_type: type[BaseException] \| None,
720	exc_val: BaseException \| None,
721	exc_tb: TracebackType \| None,
722	) -> bool \| None:
723	self._guarded = False	11✔
724	return None	11✔

agronholm / anyio / 10708046317

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