asyncutils._internal.helpers

Miscellaneous helper routines for asyncutils submodules that are not meant to be called by the user.

Classes

Bag	A thin dictionary subclass that supports attribute access, setting and deleting.
LoopMixinBase	The base class for LoopContextMixin.

Functions

check(→ bool)	Check if two objects are equal without calling the __eq__() method of the candidate, such that it cannot falsely claim equality.
check_methods(→ bool)	Re-implement collections.abc._check_methods().
coerce_callable(…)	Return the callable itself if the argument is callable, and return its type otherwise.
copy_and_clear(→ T)	Copy the given object, clear it, and return the copy.
create_executor(→ asyncutils.config.Executor)	Return an Executor instance for the given object, creating one if necessary, and saving it on the object as the executor attribute if save=False was not passed.
filter_out(→ types.GeneratorType[Any])	Yield items in the positional arguments not identical to the sentinel s, or None by default.
fullname(→ str)	Return the fully-qualified name of the given object, including its module, optionally removing the 'asyncutils' prefix.
get_loop_and_set(→ asyncio.AbstractEventLoop)	Return the running event loop. If there is none, create and set one first.
ismodule(→ TypeIs[types.ModuleType])	Whether the given object is a module, or an asyncutils-internal submodule object.
simple_wrap(→ T)	Return a coroutine wrapping the given awaitable. Use wrap_in_coro() instead for better error handling.
subscriptable(...)	Add a __class_getitem__() method to the given class, raising TypeError if it is already present.

Module Contents

class asyncutils._internal.helpers.Bag[T]

Bases: dict[str, T]

A thin dictionary subclass that supports attribute access, setting and deleting.

Initialize self. See help(type(self)) for accurate signature.

__delattr__(key: str, /) → None

Implement delattr(self, name).

__getattr__(key: str, /) → T

__setattr__(key: str, value: T, /) → None

Implement setattr(self, name, value).

class asyncutils._internal.helpers.LoopMixinBase

The base class for LoopContextMixin.

make[T](aw: collections.abc.Awaitable[T], /) → asyncio.Task[T]

Create a Task for the given awaitable that runs in the underlying loop.

make_fut() → asyncio.Future[Any]

Create a Future attached to the underlying loop.

make_multiple[T](aws: collections.abc.Iterable[collections.abc.Awaitable[T]], /) → types.GeneratorType[asyncio.Task[T]]

Return an iterator over instances of Task created for each coroutine in C, in that order.

property loop: asyncio.AbstractEventLoop

The underlying event loop.

asyncutils._internal.helpers.check(candidate: object, against: object, /) → bool

Check if two objects are equal without calling the __eq__() method of the candidate, such that it cannot falsely claim equality.

asyncutils._internal.helpers.check_methods(obj: object, /, *meth: str) → bool

Re-implement collections.abc._check_methods().

asyncutils._internal.helpers.coerce_callable[T: collections.abc.Callable[Ellipsis, Any]](f: T, /) → T

asyncutils._internal.helpers.coerce_callable(o: T, /) → type[T]

Return the callable itself if the argument is callable, and return its type otherwise.

asyncutils._internal.helpers.copy_and_clear[T: asyncutils._internal.prots.CanClearAndCopy[Any]](l: T) → T

Copy the given object, clear it, and return the copy.

asyncutils._internal.helpers.create_executor(obj: object, /, save: bool = ...) → asyncutils.config.Executor

Return an Executor instance for the given object, creating one if necessary, and saving it on the object as the executor attribute if save=False was not passed.

asyncutils._internal.helpers.filter_out(*a: object, s: object = ...) → types.GeneratorType[Any]

Yield items in the positional arguments not identical to the sentinel s, or None by default.

asyncutils._internal.helpers.fullname(f: object, /, remove_prefix: bool = ...) → str

Return the fully-qualified name of the given object, including its module, optionally removing the 'asyncutils' prefix.

asyncutils._internal.helpers.get_loop_and_set() → asyncio.AbstractEventLoop

Return the running event loop. If there is none, create and set one first.

asyncutils._internal.helpers.ismodule(o: object, /) → TypeIs[types.ModuleType]

Whether the given object is a module, or an asyncutils-internal submodule object.

async asyncutils._internal.helpers.simple_wrap[T](aw: collections.abc.Awaitable[T], /) → T

Return a coroutine wrapping the given awaitable. Use wrap_in_coro() instead for better error handling.

asyncutils._internal.helpers.subscriptable[T](cls: type[T], /) → asyncutils._internal.prots.Subscriptable[T]

Add a __class_getitem__() method to the given class, raising TypeError if it is already present.