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.

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>

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

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

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

__aiter__(): Return the {cls} for aiteration.

__anext__()

Asynchronously fetch the next item from the {cls}.

Raises:: StopAsyncIteration – Once all {obj} have been fetched from the {cls}.

__init__()

Initializes the {cls} with a function and an iterable.

Parameters:

function – A function to apply to the items in the iterable.
iterable – An iterable or an async iterable yielding objects to which function will be applied.

__iter__(): Return the {cls} for iteration.

Note

Synchronous iteration uses asyncio.BaseEventLoop.run_until_complete() to fetch {obj}. 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.

__next__()

Synchronously fetch the next item from the {cls}.

Note

This method uses asyncio.BaseEventLoop.run_until_complete() to fetch {obj}. 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:

StopIteration – Once all {obj} have been fetched from the {cls}.
SyncModeInAsyncContextError – If the event loop is already running.

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])

Filters the {obj} yielded by the {cls} based on a function.

Parameters:: function (ViewFn[T]) – 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 {obj} from the {cls}.

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the {obj} yielded by the {cls}.

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 {obj} yielded from this {cls}, but sorted.

classmethod wrap(cls, wrapped): Class method to wrap either an AsyncIterator or an async generator function.

_anext

_function

materialized

List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.
Type:: _AwaitableAsyncIterableMixin.materialized

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. .. rubric:: 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)

Welcome to ez-a-sync’s documentation!

Indices and tables