Welcome to ez-a-sync’s documentation!

ez-a-sync is a big library, but there are only a few things you need to know.

There are two main entrypoints to ez-a-sync, the a_sync decorator and the ASyncGenericBase base class. The vast majority of the other objects in this library support these two entrypoints internally. There is some more you can do, I’ll document that stuff later. For now, you have this:

@a_sync.a_sync(coro_fn=None, default=None, **modifiers)[source]

A versatile decorator that enables both synchronous and asynchronous execution of functions.

This decorator allows a function to be called either synchronously or asynchronously, depending on the context and parameters. It provides a powerful way to write code that can be used in both synchronous and asynchronous environments.

Parameters:

  • coro_fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T] | None) – The function to be decorated. Can be either a coroutine function or a regular function.

  • default (Literal['sync', 'async', None] | None) – Determines the default execution mode. Can be ‘async’, ‘sync’, or None. If None, the mode is inferred from the decorated function type.

  • **modifiers (Unpack[ModifierKwargs]) – Additional keyword arguments to modify the behavior of the decorated function. See ModifierKwargs for available options.

Return type:

ASyncDecorator | ASyncFunction[~P, T]

Modifiers:

The following modifiers can be used to customize the behavior of the decorator:

  • cache_type: Can be None or ‘memory’. ‘memory’ is an LRU cache which can be modified with the ‘cache_typed’, ‘ram_cache_maxsize’, and ‘ram_cache_ttl’ modifiers.

  • cache_typed: Set to True if you want types considered for cache keys. For example, with cache_typed=True, Decimal(0) and 0 will be considered separate keys.

  • ram_cache_maxsize: The max size for your LRU cache. None if the cache is unbounded. If you set this value without specifying a cache type, ‘memory’ will automatically be applied.

  • ram_cache_ttl: The TTL for items in your LRU cache. Set to None. If you set this value without specifying a cache type, ‘memory’ will automatically be applied.

  • runs_per_minute: Setting this value enables a rate limiter for the decorated function.

  • semaphore: Drop in a Semaphore for your async defined functions.

  • executor: The executor for the synchronous function. Set to the library’s default of config.default_sync_executor.

Examples

The decorator can be used in several ways.

  1. As a simple decorator:
    >>> @a_sync
... async def some_async_fn():
...     return True
>>> await some_async_fn()
True
>>> some_async_fn(sync=True)
True

    >>> @a_sync
... def some_sync_fn():
...     return True
>>> some_sync_fn()
True
>>> some_sync_fn(sync=False)
<coroutine object some_sync_fn at 0x7fb4f5fb49c0>
  2. As a decorator with default mode specified:
    >>> @a_sync(default='sync')
... async def some_fn():
...     return True
...
>>> some_fn()
True
>>> some_fn(sync=False)
<coroutine object some_fn at 0x7fb4f5fb49c0>

    >>> @a_sync('async')
... def some_fn():
...     return True
...
>>> some_fn()
<coroutine object some_fn at 0x7fb4f5fb49c0>
>>> some_fn(asynchronous=False)
True
  3. As a decorator with modifiers:
    >>> @a_sync(cache_type='memory', runs_per_minute=60)
... async def some_fn():
...    return True
...
>>> some_fn(sync=True)
True
  4. Applied directly to a function:
    >>> some_fn = a_sync(some_existing_function, default='sync')
>>> some_fn()
"some return value"
The decorated function can then be called either synchronously or asynchronously:
>>> result = some_fn()  # Synchronous call
>>> result = await some_fn()  # Asynchronous call
The execution mode can also be explicitly specified during the call:
>>> result = some_fn(sync=True)  # Force synchronous execution
>>> result = await some_fn(sync=False)  # Force asynchronous execution

This decorator is particularly useful for libraries that need to support both synchronous and asynchronous usage, or for gradually migrating synchronous code to asynchronous without breaking existing interfaces.

Note

If the coro_fn argument is passed as ‘async’ or ‘sync’, it is treated as the default argument, and coro_fn is set to None.

See also

ASyncFunction, ASyncDecorator

class a_sync.ASyncGenericBase

Base class for creating dual-function sync/async-capable classes without writing all your code twice.

This class, via its inherited metaclass :class:`~ASyncMeta’, provides the foundation for creating hybrid sync/async classes. It allows methods and properties to be defined once and used in both synchronous and asynchronous contexts.

The class uses the a_sync() decorator internally to create dual-mode methods and properties. Subclasses should define their methods as coroutines (using async def) where possible, and use the @a_sync.property or @a_sync.cached_property decorators for properties that need to support both modes.

Example

```python class MyClass(ASyncGenericBase):

def __init__(self, sync: bool):

self.sync = sync

@a_sync.property async def my_property(self):

return await some_async_operation()

@a_sync async def my_method(self):

return await another_async_operation()

# Synchronous usage obj = MyClass(sync=True) sync_result = obj.my_property sync_method_result = obj.my_method()

# Asynchronous usage obj = MyClass(sync=False) async_result = await obj.my_property async_method_result = await obj.my_method() ```

Note

When subclassing, be aware that all async methods and properties will be automatically wrapped to support both sync and async calls. This allows for seamless usage in different contexts without changing the underlying implementation.

__init__(self)

Those objects will work for most use cases. If you need to do some iteration, the a_sync.iter submodule has what you need:

class a_sync.iter.ASyncFilter

An async filter class that filters items of an async iterable based on a provided function.

This class inherits from _ASyncView and provides the functionality to asynchronously iterate over items, applying the filter function to each item to determine if it should be included in the result. The filter function can be either synchronous or asynchronous.

Example:
>>> async def is_even(x):
...     return x % 2 == 0
>>> filtered_iterable = ASyncFilter(is_even, some_async_iterable)
>>> async for item in filtered_iterable:
...     print(item)
See Also:

When awaited, a list of all T objects will be returned.

Example

>>> my_object = ASyncFilter(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], T)
True
__aiter__(self) → Self

Return the ASyncIterator for aiteration.

Return type:

Self

async __anext__(self) → T
Return type:

T

__init__(self, function: ViewFn[T], iterable: AnyIterable[T]) → None

Initializes the _ASyncView with a function and an iterable.

Parameters:
Return type:

None

__iter__(self) → Self

Return the ASyncIterator for iteration.

Note

Synchronous iteration uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Self

__next__(self) → T

Synchronously fetch the next item from the ASyncIterator.

Note

This method uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Raises:
Return type:

T

async _check(self, obj: T) → bool

Checks if an object passes the filter function.

Parameters:

obj (T) – The object to check.

Returns:

True if the object passes the filter, False otherwise.

Return type:

bool

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.

Returns:

An instance of ASyncFilter that yields the filtered T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by the ASyncIterable.

Parameters:

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped)

Class method to wrap either an AsyncIterator or an async generator function.

property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

class a_sync.iter.ASyncGeneratorFunction

Encapsulates an asynchronous generator function, providing a mechanism to use it as an asynchronous iterator with enhanced capabilities. This class wraps an async generator function, allowing it to be called with parameters and return an ASyncIterator object. It is particularly useful for situations where an async generator function needs to be used in a manner that is consistent with both synchronous and asynchronous execution contexts.

The ASyncGeneratorFunction class supports dynamic binding to instances, enabling it to be used as a method on class instances. When accessed as a descriptor, it automatically handles the binding to the instance, thereby allowing the wrapped async generator function to be invoked with instance context (‘self’) automatically provided. This feature is invaluable for designing classes that need to expose asynchronous generators as part of their interface while maintaining the ease of use and calling semantics similar to regular methods.

By providing a unified interface to asynchronous generator functions, this class facilitates the creation of APIs that are flexible and easy to use in a wide range of asynchronous programming scenarios. It abstracts away the complexities involved in managing asynchronous generator lifecycles and invocation semantics, making it easier for developers to integrate asynchronous iteration patterns into their applications.

Example

>>> async def my_async_gen():
...     yield 1
...     yield 2
>>> async_gen_func = ASyncGeneratorFunction(my_async_gen)
>>> for item in async_gen_func():
...     print(item)

See also

__call__(self, *args: P.args, **kwargs: P.kwargs) → ASyncIterator[T]

Calls the wrapped async generator function with the given arguments and keyword arguments, returning an ASyncIterator.

Parameters:

  • *args (~P) – Positional arguments for the function.

  • **kwargs (~P) – Keyword arguments for the function.

Return type:

ASyncIterator[T]

__init__(self, async_gen_func: AsyncGenFunc[P, T], instance: Any = None) → None

Initializes the ASyncGeneratorFunction with the given async generator function and optionally an instance.

Parameters:

  • async_gen_func (Callable[[~P], AsyncGenerator[T, None] | AsyncIterator[T]]) – The async generator function to wrap.

  • instance (optional) – The object to bind to the function, if applicable.

Return type:

None

_cache_handle: TimerHandle
class a_sync.iter.ASyncIterable

A hybrid Iterable/AsyncIterable implementation designed to offer dual compatibility with both synchronous and asynchronous iteration protocols.

This class allows objects to be iterated over using either a standard for loop or an async for loop, making it versatile in scenarios where the mode of iteration (synchronous or asynchronous) needs to be flexible or is determined at runtime.

The class achieves this by implementing both __iter__ and __aiter__ methods, enabling it to return appropriate iterator objects that can handle synchronous and asynchronous iteration, respectively. However, note that synchronous iteration relies on the ASyncIterator class, which uses asyncio.get_event_loop().run_until_complete to fetch items. This can raise a RuntimeError if the event loop is already running, and in such cases, a SyncModeInAsyncContextError is raised from the RuntimeError.

Example:
>>> async_iterable = ASyncIterable(some_async_iterable)
>>> async for item in async_iterable:
...     print(item)
>>> for item in async_iterable:
...     print(item)
See Also:

When awaited, a list of all T objects will be returned.

Example

>>> my_object = ASyncIterable(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], T)
True
__aiter__(self) → AsyncIterator[T]

Return an async iterator that yields T objects from the ASyncIterable.

Return type:

AsyncIterator[T]

__init__(self, async_iterable: AsyncIterable[T])

Initializes the ASyncIterable with an async iterable.

Parameters:

async_iterable (AsyncIterable[T]) – The async iterable to wrap.

__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the ASyncIterable.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Iterator[T]

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.

Returns:

An instance of ASyncFilter that yields the filtered T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by the ASyncIterable.

Parameters:

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:

wrapped (AsyncIterable[T])

Return type:

ASyncIterable[T]

property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

class a_sync.iter.ASyncIterator

A hybrid Iterator/AsyncIterator implementation that bridges the gap between synchronous and asynchronous iteration. This class provides a unified interface for iteration that can seamlessly operate in both synchronous (for loop) and asynchronous (async for loop) contexts. It allows the wrapping of asynchronous iterable objects or async generator functions, making them usable in synchronous code without explicitly managing event loops or asynchronous context switches.

By implementing both __next__ and __anext__ methods, ASyncIterator enables objects to be iterated using standard iteration protocols while internally managing the complexities of asynchronous iteration. This design simplifies the use of asynchronous iterables in environments or frameworks that are not inherently asynchronous, such as standard synchronous functions or older codebases being gradually migrated to asynchronous IO.

Note:

Synchronous iteration with ASyncIterator uses asyncio.get_event_loop().run_until_complete, which can raise a RuntimeError if the event loop is already running. In such cases, a SyncModeInAsyncContextError is raised from the RuntimeError, indicating that synchronous iteration is not possible in an already running event loop.

Example:
>>> async_iterator = ASyncIterator(some_async_iterator)
>>> async for item in async_iterator:
...     print(item)
>>> for item in async_iterator:
...     print(item)
See Also:

When awaited, a list of all T objects will be returned.

Example

>>> my_object = ASyncIterator(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], T)
True
__aiter__(self) → Self

Return the ASyncIterator for aiteration.

Return type:

Self

__anext__(self) → Coroutine[Any, Any, T]

Asynchronously fetch the next item from the ASyncIterator.

Raises:

StopAsyncIteration – Once all T objects have been fetched from the ASyncIterator.

Return type:

Coroutine[Any, Any, T]

__init__(self, async_iterator: AsyncIterator[T])

Initializes the ASyncIterator with an async iterator.

Parameters:

async_iterator (AsyncIterator[T]) – The async iterator to wrap.

__iter__(self) → Self

Return the ASyncIterator for iteration.

Note

Synchronous iteration uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Self

__next__(self) → T

Synchronously fetch the next item from the ASyncIterator.

Note

This method uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Raises:
Return type:

T

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.

Returns:

An instance of ASyncFilter that yields the filtered T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by the ASyncIterable.

Parameters:

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped)

Class method to wrap either an AsyncIterator or an async generator function.

property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

class a_sync.iter.ASyncSorter

An async sorter class that sorts items of an async iterable based on a provided key function.

This class inherits from _ASyncView and provides the functionality to asynchronously iterate over items, applying the key function to each item for sorting. The key function can be either synchronous or asynchronous. Note that the ASyncSorter instance can only be consumed once.

Example:
>>> sorted_iterable = ASyncSorter(some_async_iterable, key=lambda x: x.value)
>>> async for item in sorted_iterable:
...     print(item)
See Also:

When awaited, a list of all T objects will be returned.

Example

>>> my_object = ASyncSorter(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], T)
True
__aiter__(self) → AsyncIterator[T]

Return an async iterator for the ASyncSorter.

Raises:

RuntimeError – If the ASyncSorter instance has already been consumed.

Returns:

An async iterator that will yield the sorted T objects.

Return type:

AsyncIterator[T]

__anext__(self) → T
Return type:

T

__init__(self, iterable: AsyncIterable[T], *, key: SortKey[T] = None, reverse: bool = False) → None

Initializes the ASyncSorter with an iterable and an optional sorting configuration (key function, and reverse flag).

Parameters:

  • iterable (AsyncIterable[T]) – The async iterable to sort.

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If none is provided, elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the list elements will be sorted in reverse order. Defaults to False.

Return type:

None

__iter__(self) → Self

Return the ASyncIterator for iteration.

Note

Synchronous iteration uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:

Self

__next__(self) → T

Synchronously fetch the next item from the ASyncIterator.

Note

This method uses asyncio.BaseEventLoop.run_until_complete() to fetch T objects. This raises a RuntimeError if the event loop is already running. This RuntimeError will be caught and a more descriptive SyncModeInAsyncContextError will be raised in its place.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Raises:
Return type:

T

filter(self, function: ViewFn[T]) → 'ASyncFilter[T]'

Filters the T objects yielded by the ASyncIterable based on a function.

Parameters:

function (Callable[[T], Awaitable[bool]] | Callable[[T], bool]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.

Returns:

An instance of ASyncFilter that yields the filtered T objects from the ASyncIterable.

Return type:

ASyncFilter[T]

sort(self, *, key: SortKey[T] = None, reverse: bool = False) → 'ASyncSorter[T]'

Sort the T objects yielded by the ASyncIterable.

Parameters:

  • key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.

  • reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this ASyncIterable, but sorted.

Return type:

ASyncSorter[T]

classmethod wrap(cls, wrapped)

Class method to wrap either an AsyncIterator or an async generator function.

_consumed: bool = False
property materialized: List[T]

Synchronously iterate through the ASyncIterable and return all T objects.

Returns:

A list of the T objects yielded by the ASyncIterable.

reversed: bool = False

There’s this cool future class that doesn’t interact with the rest of the lib but, depending on your needs, might be even better. WIP:

Everything else in ez-a-sync can be found by navigating the tree below. Enjoy!

Contents:

Indices and tables