asyncutils.mixins

Mixins classes for some common or specialized patterns that provide methods based on some abstract methods.

Classes

AsyncContextMixin
AwaitableMixin	A subclass that implements the async wait() method automatically becomes awaitable, resolving to the return value of that method.
EventMixin
ExecutorRequiredAsyncContextMixin	Version of AsyncContextMixin that uses an executor to convert the sync context manager methods to async in an event loop.
LockMixin	A mixin to derive __aenter__() and __aexit__() from acquire() and release() of subclasses.
LockWithOwnerMixin	Mixin for locks that can report their owner (the task currently holding it).
LoopContextMixin	Like asyncio.TaskGroup, but manages an event loop publicly, allows custom setup and teardown logic and waits for the cancellations to complete on exit.

Module Contents

class asyncutils.mixins.AsyncContextMixin[T]

Bases: abc.ABC

A mixin to derive __aenter__() and __aexit__() from __enter__() and __exit__() of subclasses.

__enter__() is optional and returns self by default, but that cannot be typed accurately.

async __aenter__() → T

async __aexit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → bool | None

async __aexit__(exc_typ: None, exc_val: None, exc_tb: None, /) → Literal[False] | None

__enter__() → T

abstractmethod __exit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → bool | None

abstractmethod __exit__(exc_typ: None, exc_val: None, exc_tb: None, /) → Literal[False] | None

class asyncutils.mixins.AwaitableMixin[T]

Bases: abc.ABC

A subclass that implements the async wait() method automatically becomes awaitable, resolving to the return value of that method.

__await__() → collections.abc.Generator[Any, None, T]

Await statement support.

abstractmethod wait() → collections.abc.Awaitable[T]

class asyncutils.mixins.EventMixin[T]

Bases: AwaitableMixin[T], asyncutils._internal.helpers.LoopMixinBase, abc.ABC

Mixin for event classes that don’t inherit from asyncio.Event but provide enhanced functionality with the same API and some mixin

methods, most notably making the event itself awaitable. This is simply syntactic sugar for calling the wait method, but more convenient and

intuitive when thinking of events as reusable futures.

abstractmethod clear() → None

Clear the value of the event.

abstractmethod get() → T

Return the value currently held by the event, or raise an exception if there is no value set. The implementations in this library have a dedicated exception type, EventValueError, for this.

abstractmethod is_set() → bool

Return whether the event currently holds a value.

abstractmethod set(value: T) → None

Set the value of the event and wake up all waiters. Implementations may not choose to clear the value within this method.

async wait(timeout: float | None = ...) → T

Wait for the event with an optional timeout and return its value.

abstractmethod wait_for_next(timeout: float | None = ...) → T

Async:

Wait for the next time the event is set, and return the value it was set to. Should always block even if there is currently a value set.

async wait_for_value(val: T, timeout: float | None = ..., *, set_at_timeout: bool = ...) → None

Wait for the event to be set to a specific value val, with an optional timeout. If set_at_timeout is True, the event will be set to val when the timeout occurs, and waiters will be woken up. The value will persist on the event by default in this case.

class asyncutils.mixins.ExecutorRequiredAsyncContextMixin[T]

Bases: abc.ABC

Version of AsyncContextMixin that uses an executor to convert the sync context manager methods to async in an event loop.

async __aenter__() → T

async __aexit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → bool | None

async __aexit__(exc_typ: None, exc_val: None, exc_tb: None, /) → Literal[False] | None

__enter__() → T

abstractmethod __exit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → bool | None

abstractmethod __exit__(exc_typ: None, exc_val: None, exc_tb: None, /) → Literal[False] | None

property runner: collections.abc.Callable[[collections.abc.Callable[[*Ts], R], *Ts], asyncio.Future[R]]

class asyncutils.mixins.LockMixin[T]

Bases: abc.ABC

A mixin to derive __aenter__() and __aexit__() from acquire() and release() of subclasses.

async __aenter__() → T

Acquire the lock and return an object of the implementation’s choice.

async __aexit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → None

async __aexit__(exc_typ: None, exc_val: None, exc_tb: None, /) → None

Release the lock and optionally perform handling according to the exception occurred.

classmethod __init_subclass__(*, _lock_factory_: collections.abc.Callable[[Self], T] = ..., **k: object) → None

acknowledge_locksmith_lock_held(

smith: asyncutils.locksmiths.LocksmithBase,

/,

) → bool | collections.abc.Coroutine[Any, Any, bool]

Optional method to cooperate with locksmiths in case of unusually long wait times for the lock which may indicate deadlock, livelock or starvation.

abstractmethod acquire() → bool

Async:

Acquire the lock, waiting if necessary. Return whether the lock was acquired successfully. Some locks may raise an error if not; that is up to the implementation.

abstractmethod locked() → bool

Return whether the lock is currently held by a task satisfying a certain criterion (e.g. it is the task awaited the acquiry).

abstractmethod release() → collections.abc.Coroutine[Any, Any, None] | None

Release the lock. Some locks may require this to be called from a task that holds the lock, or may raise an error if the lock is not currently held; that is up to the implementation.

class asyncutils.mixins.LockWithOwnerMixin[R: (None, collections.abc.Coroutine[Any, Any, None])]

Bases: LockMixin[None]

Mixin for locks that can report their owner (the task currently holding it).

abstractmethod _release() → R

Will be wrapped by release() to throw RuntimeError if the current task is not the owner of the lock.

release() → R

Release the lock. Some locks may require this to be called from a task that holds the lock, or may raise an error if the lock is not currently held; that is up to the implementation.

property is_owner: bool

Abstractmethod:

Should evaluate to True if the current task is the owner of the lock

class asyncutils.mixins.LoopContextMixin

Bases: asyncutils._internal.helpers.LoopMixinBase

Like asyncio.TaskGroup, but manages an event loop publicly, allows custom setup and teardown logic and waits for the cancellations to complete on exit.

async __aenter__() → Self

Call __setup__() and return self.

async __aexit__(exc_typ: asyncutils._internal.types.ExcType, exc_val: BaseException, exc_tb: types.TracebackType, /) → None

async __aexit__(exc_typ: None, exc_val: None, exc_tb: None, /) → None

Call __cleanup__() and cancel all running tasks in the loop.

async __cleanup__() → None

Default implementation does nothing.

async __setup__() → None

Default implementation does nothing.

property loop: asyncio.AbstractEventLoop

The underlying event loop.

property running_tasks: set[asyncio.Task[Any]]

A set of all tasks currently running in the underlying loop.