asyncutils.futures¶

Various implementations of future and task classes, eager, time-aware and supporting asynchronous and no-argument callbacks.

Classes¶

`AsyncCallbacksFuture`	A subclass of `Future` that supports calling asynchronous callbacks and callbacks with no arguments on completion.
`AsyncCallbacksTask`	Self-explanatory.
`TimeAwareAsyncCallbacksFuture`	A subclass of `AsyncCallbacksFuture` that can be compared to other `TimeAwareAsyncCallbacksFuture`'s based on the time they were created.
`TimeAwareAsyncCallbacksTask`	A subclass of `AsyncCallbacksTask` that can be compared to other `TimeAwareAsyncCallbacksTask`'s based on the time they were created.
`TimeAwareFuture`	A subclass of `Future` that can be compared to other `TimeAwareFuture`'s based on the time they were created.
`TimeAwareTask`	A subclass of `Task` that can be compared to other `TimeAwareTask`'s based on the time they were created.
`TimeAwareUniqueCallbacksFuture`	A subclass of `UniqueCallbacksFuture` that can be compared to other `TimeAwareUniqueCallbacksFuture`'s based on the time they were created.
`TimeAwareUniqueCallbacksTask`	A subclass of `UniqueCallbacksTask` that can be compared to other `TimeAwareUniqueCallbacksTask`'s based on the time they were created.
`UniqueCallbacksFuture`	Like `AsyncCallbacksFuture`, but disallow the same callback from being added twice. Removal is faster and more intuitive to use.
`UniqueCallbacksTask`	Self-explanatory.

Module Contents¶

class asyncutils.futures.AsyncCallbacksFuture[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: asyncio.Future[T]

A subclass of Future that supports calling asynchronous callbacks and callbacks with no arguments on completion.

Note

To hook into the callbacks mechanism, subclassing the C-accelerated implementation of Future is impossible; i.e., using many of them, for example when implementing a queue, may be slower.

add_async_callback( fn: collections.abc.Callable[[Self], collections.abc.Awaitable[object]], /, *, context: contextvars.Context | None = ..., ) → None¶: Add an asynchronous callback to be called when the future is done. The callback will be passed the future as an argument.

add_noargs_async_callback( fn: collections.abc.Callable[[], collections.abc.Awaitable[object]], /, *, context: contextvars.Context | None = ..., ) → None¶: Add an asynchronous callback with no arguments to be called when the future is done.

add_noargs_callback(fn: collections.abc.Callable[[], object], /, *, context: contextvars.Context | None = ...) → None¶: Add a callback with no arguments to be called when the future is done.

remove_async_callback(fn: collections.abc.Callable[[Self], collections.abc.Awaitable[object]], /) → int¶: Remove an asynchronous callback. Returns the number of callbacks removed.

remove_done_callback(fn: collections.abc.Callable[[Self], object], /) → int¶: Remove a callback. Returns the number of callbacks removed.

remove_noargs_async_callback(fn: collections.abc.Callable[[], collections.abc.Awaitable[object]], /) → int¶: Remove an asynchronous callback with no arguments. Returns the number of callbacks removed.

remove_noargs_callback(fn: collections.abc.Callable[[], object], /) → int¶: Remove a callback with no arguments. Returns the number of callbacks removed.

class asyncutils.futures.AsyncCallbacksTask[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: asyncio.Task[T], AsyncCallbacksFuture[T]

Self-explanatory.

class asyncutils.futures.TimeAwareAsyncCallbacksFuture[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: TimeAwareFuture[T], AsyncCallbacksFuture[T]

A subclass of AsyncCallbacksFuture that can be compared to other TimeAwareAsyncCallbacksFuture’s based on the time they were created.

class asyncutils.futures.TimeAwareAsyncCallbacksTask[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: TimeAwareTask[T], AsyncCallbacksTask[T]

A subclass of AsyncCallbacksTask that can be compared to other TimeAwareAsyncCallbacksTask’s based on the time they were created.

class asyncutils.futures.TimeAwareFuture[T][source]¶

Bases: asyncio.Future[T]

A subclass of Future that can be compared to other TimeAwareFuture’s based on the time they were created.

__lt__(other: TimeAwareFuture[Any] | TimeAwareTask[Any], /) → bool¶

class asyncutils.futures.TimeAwareTask[T][source]¶

Bases: asyncio.Task[T]

A subclass of Task that can be compared to other TimeAwareTask’s based on the time they were created.

__lt__(other: TimeAwareFuture[Any] | TimeAwareTask[Any], /) → bool¶

class asyncutils.futures.TimeAwareUniqueCallbacksFuture[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: TimeAwareFuture[T], UniqueCallbacksFuture[T]

A subclass of UniqueCallbacksFuture that can be compared to other TimeAwareUniqueCallbacksFuture’s based on the time they were created.

class asyncutils.futures.TimeAwareUniqueCallbacksTask[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: TimeAwareTask[T], UniqueCallbacksTask[T]

A subclass of UniqueCallbacksTask that can be compared to other TimeAwareUniqueCallbacksTask’s based on the time they were created.

class asyncutils.futures.UniqueCallbacksFuture[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: asyncio.Future[T]

Like AsyncCallbacksFuture, but disallow the same callback from being added twice. Removal is faster and more intuitive to use.

add_async_callback( fn: collections.abc.Callable[[Self], collections.abc.Awaitable[object]], /, *, context: contextvars.Context | None = ..., ) → None¶: Add an asynchronous callback to be called when the future is done. The callback will be passed the future as an argument.

add_done_callback(fn: collections.abc.Callable[[Self], object], /, *, context: contextvars.Context | None = ...) → None¶: Add a callback to be called when the future is done. The callback will be passed the future as an argument.

add_noargs_async_callback( fn: collections.abc.Callable[[], collections.abc.Awaitable[object]], /, *, context: contextvars.Context | None = ..., ) → None¶: Add an asynchronous callback with no arguments to be called when the future is done.

add_noargs_callback(fn: collections.abc.Callable[[], object], /, *, context: contextvars.Context | None = ...) → None¶: Add a callback with no arguments to be called when the future is done.

remove_async_callback(fn: collections.abc.Callable[[Self], collections.abc.Awaitable[object]], /) → Literal[0, 1]¶: Remove an asynchronous callback. Returns the number of callbacks removed.

remove_done_callback(fn: collections.abc.Callable[[Self], object], /) → Literal[0, 1]¶: Remove a callback. Returns the number of callbacks removed.

remove_noargs_async_callback(fn: collections.abc.Callable[[], collections.abc.Awaitable[object]], /) → Literal[0, 1]¶: Remove an asynchronous callback with no arguments. Returns the number of callbacks removed.

remove_noargs_callback(fn: collections.abc.Callable[[], object], /) → Literal[0, 1]¶: Remove a callback with no arguments. Returns the number of callbacks removed.

class asyncutils.futures.UniqueCallbacksTask[T](*, loop: asyncio.AbstractEventLoop | None = ...)[source]¶

Bases: asyncio.Task[T], UniqueCallbacksFuture[T]

Self-explanatory.