a_sync.a_sync package
Subpackages
- a_sync.a_sync.modifiers package
- Subpackages
- Submodules
- a_sync.a_sync.modifiers.limiter module
- a_sync.a_sync.modifiers.manager module
ModifierManagerModifierManager.__getitem__()ModifierManager.__init__()ModifierManager.__iter__()ModifierManager.apply_async_modifiers()ModifierManager.apply_sync_modifiers()ModifierManager.clear()ModifierManager.copy()ModifierManager.fromkeys()ModifierManager.get()ModifierManager.items()ModifierManager.keys()ModifierManager.pop()ModifierManager.popitem()ModifierManager.setdefault()ModifierManager.update()ModifierManager.values()ModifierManager._modifiersModifierManager.cache_typeModifierManager.cache_typedModifierManager.defaultModifierManager.executorModifierManager.ram_cache_maxsizeModifierManager.ram_cache_ttlModifierManager.runs_per_minuteModifierManager.semaphoreModifierManager.use_cacheModifierManager.use_limiterModifierManager.use_semaphore
- a_sync.a_sync.modifiers.semaphores module
- Module contents
Submodules
a_sync.a_sync._descriptor module
This module contains the ASyncDescriptor class, which is used to create sync/async methods and properties.
It also includes utility methods for mapping operations across multiple instances.
- class a_sync.a_sync._descriptor.ASyncDescriptor[source]
Bases:
ModifiedMixin,Generic[I,P,T]A descriptor base class for asynchronous methods and properties.
This class provides functionality for mapping operations across multiple instances and includes utility methods for common operations.
- __init__(_fget, field_name=None, **modifiers)[source]
Initialize the {cls}.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (ModifierKwargs) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)[source]
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)[source]
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- map(*instances, **bound_method_kwargs)[source]
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
a_sync.a_sync._flags module
This module provides functionality for handling synchronous and asynchronous flags in the ez-a-sync library.
ez-a-sync uses ‘flags’ to indicate whether objects / function calls will be sync or async.
You can use any of the provided flags, whichever makes most sense for your use case.
- a_sync.a_sync._flags.negate_if_necessary(flag, flag_value)[source]
Negate the flag value if necessary based on the flag type.
- Parameters:
- Returns:
The potentially negated flag value.
- Raises:
exceptions.InvalidFlag – If the flag is not recognized.
- Return type:
- a_sync.a_sync._flags.validate_flag_value(flag, flag_value)[source]
Validate that the flag value is a boolean.
- Parameters:
- Returns:
The validated flag value.
- Raises:
exceptions.InvalidFlagValue – If the flag value is not a boolean.
- Return type:
- a_sync.a_sync._flags.AFFIRMATIVE_FLAGS = {'sync'}
Set of flags indicating synchronous behavior.
- a_sync.a_sync._flags.NEGATIVE_FLAGS = {'asynchronous'}
Set of flags indicating asynchronous behavior.
- a_sync.a_sync._flags.VIABLE_FLAGS = {'asynchronous', 'sync'}
Set of all valid flags.
a_sync.a_sync._helpers module
This module provides utility functions for handling asynchronous operations and converting synchronous functions to asynchronous ones.
- a_sync.a_sync._helpers._asyncify(func, executor)[source]
Convert a synchronous function to a coroutine function.
- Parameters:
func (Callable[[~P], T]) – The synchronous function to be converted.
executor (Executor) – The executor to run the synchronous function.
- Returns:
A coroutine function wrapping the input function.
- Raises:
exceptions.FunctionNotSync – If the input function is already asynchronous.
- Return type:
- a_sync.a_sync._helpers._await(awaitable)[source]
Await an awaitable object in a synchronous context.
- Parameters:
awaitable (Awaitable[T]) – The awaitable object to be awaited.
- Returns:
The result of the awaitable.
- Raises:
exceptions.SyncModeInAsyncContextError – If the event loop is already running.
- Return type:
T
a_sync.a_sync._kwargs module
This module provides utility functions for handling keyword arguments related to synchronous and asynchronous flags.
- a_sync.a_sync._kwargs.get_flag_name(kwargs)[source]
Get the name of the flag present in the kwargs.
- Parameters:
kwargs (dict) – A dictionary of keyword arguments.
- Returns:
The name of the flag if present, None otherwise.
- Raises:
exceptions.TooManyFlags – If more than one flag is present in the kwargs.
- Return type:
str | None
a_sync.a_sync._meta module
- class a_sync.a_sync._meta.ASyncMeta[source]
Bases:
ABCMetaAny class with metaclass ASyncMeta will have its functions wrapped with a_sync upon class instantiation.
- __call__(*args, **kwargs)
Call self as a function.
- __init__(*args, **kwargs)
- _abc_caches_clear()
Clear the caches (for debugging or testing).
- _abc_registry_clear()
Clear the registry (for debugging or testing).
- _dump_registry(file=None)
Debug helper to print the ABC registry.
- mro()
Return a type’s method resolution order.
- register(subclass)
Register a virtual subclass of an ABC.
Returns the subclass, to allow usage as a class decorator.
- class a_sync.a_sync._meta.ASyncSingletonMeta[source]
Bases:
ASyncMeta- _abc_caches_clear()
Clear the caches (for debugging or testing).
- _abc_registry_clear()
Clear the registry (for debugging or testing).
- _dump_registry(file=None)
Debug helper to print the ABC registry.
- mro()
Return a type’s method resolution order.
- register(subclass)
Register a virtual subclass of an ABC.
Returns the subclass, to allow usage as a class decorator.
a_sync.a_sync.abstract module
a_sync.a_sync.base module
- class a_sync.a_sync.base.ASyncGenericBase[source]
Bases:
ASyncABCBase class for creating dual-function sync/async-capable classes without writing all your code twice.
This class 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.
a_sync.a_sync.config module
Configuration module for a_sync library.
This module provides configuration options and default settings for the a_sync library. It includes functionality for setting up executors, defining default modifiers, and handling environment variable configurations.
- a_sync.a_sync.config.get_default_executor()[source]
Get the default executor based on the
EXECUTOR_TYPEenvironment variable.- Returns:
An instance of either ProcessPoolExecutor or ThreadPoolExecutor.
- Return type:
Executor
- Raises:
ValueError – If an invalid EXECUTOR_TYPE is specified.
a_sync.a_sync.decorator module
- a_sync.a_sync.decorator.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
ModifierKwargsfor available options.
- Return type:
ASyncDecorator | ASyncFunction[~P, T]
- Modifiers:
- lib defaults:
- async settings
cache_type: CacheType = None, - This can be None or ‘memory’. ‘memory’ is a lru cache which can be modified with the ‘cache_typed’,’ram_cache_maxsize’,’ram_cache_ttl’ modifiers. cache_typed: bool = False, - Set to True if you want types considered treated for cache keys. ie with cache_typed=True, Decimal(0) and 0 will be considered separate keys. ram_cache_maxsize: Optional[int] = -1, - The maxsize for your lru cache. None if cache is unbounded. If you set this value without specifying a cache type, ‘memory’ will automatically be applied. ram_cache_ttl: Optional[int] = None, - 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: Optional[int] = None, - Setting this value enables a rate limiter for the decorated function. semaphore: SemaphoreSpec = None, - drop in a Semaphore for your async defined functions.
- sync settings
executor: Executor = config.default_sync_executor
- Returns:
An
ASyncDecoratorif used as a decorator factory, or anASyncFunctionif used directly on a function.- Parameters:
coro_fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T] | None)
default (Literal['sync', 'async', None] | None)
modifiers (Unpack[ModifierKwargs])
- Return type:
ASyncDecorator | ASyncFunction[~P, T]
Examples
The decorator can be used in several ways.
1. As a simple decorator:
`python >>> @a_sync ... async def some_async_fn(): ... return True >>> await some_fn() True >>> some_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 0x12345678> `2. As a decorator with default mode specified:
`python >>> @a_sync(default='sync') ... async def some_fn(): ... return True ... >>> some_fn() True `3. As a decorator with modifiers:
`python >>> @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:
`python >>> 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:
`python result = some_fn() # Synchronous call result = await some_fn() # Asynchronous call `The execution mode can also be explicitly specified during the call:
`python 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.
a_sync.a_sync.function module
- class a_sync.a_sync.function.ASyncDecorator[source]
Bases:
ModifiedMixin- __call__(func)[source]
Call self as a function.
- Parameters:
- Return type:
ASyncFunction[~P, T]
- __init__(**modifiers)[source]
- Parameters:
modifiers (Unpack[ModifierKwargs])
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- modifiers: ModifierManager
- wrapped
- class a_sync.a_sync.function.ASyncDecoratorAsyncDefault[source]
Bases:
ASyncDecorator- __call__(func)[source]
Call self as a function.
- Parameters:
- Return type:
ASyncFunctionAsyncDefault[~P, T]
- __init__(**modifiers)
- Parameters:
modifiers (Unpack[ModifierKwargs])
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- validate_inputs()
- Return type:
None
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- modifiers: ModifierManager
- wrapped
- class a_sync.a_sync.function.ASyncDecoratorSyncDefault[source]
Bases:
ASyncDecorator- __call__(func)[source]
Call self as a function.
- Parameters:
- Return type:
ASyncFunctionSyncDefault[~P, T]
- __init__(**modifiers)
- Parameters:
modifiers (Unpack[ModifierKwargs])
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- validate_inputs()
- Return type:
None
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- modifiers: ModifierManager
- wrapped
- class a_sync.a_sync.function.ASyncFunction[source]
Bases:
ModifiedMixin,Generic[P,T]A callable wrapper object that can be executed both synchronously and asynchronously.
This class wraps a function or coroutine function, allowing it to be called in both synchronous and asynchronous contexts. It provides a flexible interface for handling different execution modes and applying various modifiers to the function’s behavior.
The class supports various modifiers that can alter the behavior of the function, such as caching, rate limiting, and execution in specific contexts (e.g., thread pools).
Example
```python async def my_coroutine(x: int) -> str:
return str(x)
func = ASyncFunction(my_coroutine)
# Synchronous call result = func(5, sync=True) # returns “5”
# Asynchronous call result = await func(5) # returns “5” ```
- __call__(*args, **kwargs)[source]
Call the wrapped function either synchronously or asynchronously.
This method determines whether to execute the wrapped function synchronously or asynchronously based on the default mode and any provided flags.
- Parameters:
*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.
- Returns:
The result of the wrapped function call, which may be a coroutine if run asynchronously.
- Raises:
Exception – Any exception that may be raised by the wrapped function.
- Return type:
- __init__(fn, **modifiers)[source]
Initialize an ASyncFunction instance.
- Parameters:
fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to wrap.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)[source]
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Check if all results of the function applied to the iterables are truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Check if any result of the function applied to the iterables is truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- map(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Create a TaskMapping for the wrapped function with the given iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[P, T]
- async max(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Find the maximum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The maximum result.
- Return type:
T
- async min(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Find the minimum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The minimum result.
- Return type:
T
- async sum(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Calculate the sum of the results of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The sum of the results.
- Return type:
T
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.function.ASyncFunctionAsyncDefault[source]
Bases:
ASyncFunction[P,T]A specialized
ASyncFunctionthat defaults to asynchronous execution.This class is used when the
a_sync()decorator is applied with default=’async’. It provides type hints to indicate that the default call behavior is asynchronous and supports IDE type checking for most use cases.The wrapped function can still be called synchronously by passing sync=True or asynchronous=False as a keyword argument.
Example
```python @a_sync(default=’async’) async def my_function(x: int) -> str:
return str(x)
# Asynchronous call (default behavior) result = await my_function(5) # returns “5”
# Synchronous call result = my_function(5, sync=True) # returns “5” ```
- __call__(*args, **kwargs)[source]
Call the wrapped function, defaulting to asynchronous execution.
This method overrides the base
ASyncFunction.__call__()to provide an asynchronous default behavior.- Parameters:
*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.
- Returns:
A coroutine object representing the asynchronous execution of the wrapped function.
- Raises:
Exception – Any exception that may be raised by the wrapped function.
- Return type:
- __init__(fn, **modifiers)
Initialize an ASyncFunction instance.
- Parameters:
fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to wrap.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **function_kwargs)
Check if all results of the function applied to the iterables are truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **function_kwargs)
Check if any result of the function applied to the iterables is truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- map(*iterables, concurrency=None, task_name='', **function_kwargs)
Create a TaskMapping for the wrapped function with the given iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[P, T]
- async max(*iterables, concurrency=None, task_name='', **function_kwargs)
Find the maximum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The maximum result.
- Return type:
T
- async min(*iterables, concurrency=None, task_name='', **function_kwargs)
Find the minimum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The minimum result.
- Return type:
T
- async sum(*iterables, concurrency=None, task_name='', **function_kwargs)
Calculate the sum of the results of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The sum of the results.
- Return type:
T
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.function.ASyncFunctionSyncDefault[source]
Bases:
ASyncFunction[P,T]A specialized
ASyncFunctionthat defaults to synchronous execution.This class is used when the
a_sync()decorator is applied with default=’sync’. It provides type hints to indicate that the default call behavior is synchronous and supports IDE type checking for most use cases.The wrapped function can still be called asynchronously by passing sync=False or asynchronous=True as a keyword argument.
Example
```python @a_sync(default=’sync’) async def my_function(x: int) -> str:
return str(x)
# Synchronous call (default behavior) result = my_function(5) # returns “5”
# Asynchronous call result = await my_function(5, sync=False) # returns “5” ```
- __call__(*args, **kwargs)[source]
Call the wrapped function, defaulting to synchronous execution.
This method overrides the base
ASyncFunction.__call__()to provide a synchronous default behavior.- Parameters:
*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.
- Returns:
The result of the wrapped function call.
- Raises:
Exception – Any exception that may be raised by the wrapped function.
- Return type:
- __init__(fn, **modifiers)
Initialize an ASyncFunction instance.
- Parameters:
fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to wrap.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **function_kwargs)
Check if all results of the function applied to the iterables are truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **function_kwargs)
Check if any result of the function applied to the iterables is truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- map(*iterables, concurrency=None, task_name='', **function_kwargs)
Create a TaskMapping for the wrapped function with the given iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[P, T]
- async max(*iterables, concurrency=None, task_name='', **function_kwargs)
Find the maximum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The maximum result.
- Return type:
T
- async min(*iterables, concurrency=None, task_name='', **function_kwargs)
Find the minimum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The minimum result.
- Return type:
T
- async sum(*iterables, concurrency=None, task_name='', **function_kwargs)
Calculate the sum of the results of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The sum of the results.
- Return type:
T
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.function.ModifiedMixin[source]
Bases:
objectA mixin class that provides functionality for applying modifiers to functions.
This class is used as a base for
ASyncFunctionand its variants to handle the application of async and sync modifiers to functions.- _asyncify(func)[source]
Convert a synchronous function to an asynchronous one and apply async modifiers.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- modifiers: ModifierManager
- wrapped
a_sync.a_sync.method module
This module provides classes for implementing dual-functional sync/async methods in Python.
It includes descriptors and bound methods that can be used to create flexible asynchronous interfaces, allowing methods to be called both synchronously and asynchronously based on various conditions and configurations.
- class a_sync.a_sync.method.ASyncBoundMethod[source]
Bases:
ASyncFunction[P,T],Generic[I,P,T]A bound method that can be called both synchronously and asynchronously.
This class represents a method bound to an instance, which can be called either synchronously or asynchronously based on various conditions.
- __call__(*args, **kwargs)[source]
Call the bound method.
This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.
- __cancel_cache_handle(instance)
Cancel the cache handle.
- Parameters:
instance (I) – The instance associated with the cache handle.
- Return type:
None
- __init__(instance, unbound, async_def, **modifiers)[source]
Initialize the bound method.
- Parameters:
instance (I) – The instance to bind the method to.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.
async_def (bool) – Whether the original function is an async def.
**modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **kwargs)[source]
Check if all of the results are truthy.
- Parameters:
- Returns:
True if all results are truthy, False otherwise.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **kwargs)[source]
Check if any of the results are truthy.
- Parameters:
- Returns:
True if any result is truthy, False otherwise.
- Return type:
- map(*iterables, concurrency=None, task_name='', **kwargs)[source]
Create a TaskMapping for this method.
- Parameters:
- Returns:
A TaskMapping instance for this method.
- Return type:
TaskMapping[I, T]
- async sum(*iterables, concurrency=None, task_name='', **kwargs)[source]
Calculate the sum of the results.
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- _cache_handle: TimerHandle
An asyncio handle used to pop the bound method from instance.__dict__ 5 minutes after its last use.
- _is_async_def
True if self.__wrapped__ is a coroutine function, False otherwise.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.method.ASyncBoundMethodAsyncDefault[source]
Bases:
ASyncBoundMethod[I,P,T]A bound method with asynchronous default behavior.
- __init__(instance, unbound, async_def, **modifiers)
Initialize the bound method.
- Parameters:
instance (I) – The instance to bind the method to.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.
async_def (bool) – Whether the original function is an async def.
**modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- _should_await(kwargs)
Determine if the method should be awaited.
- async all(*iterables, concurrency=None, task_name='', **kwargs)
Check if all of the results are truthy.
- Parameters:
- Returns:
True if all results are truthy, False otherwise.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **kwargs)
Check if any of the results are truthy.
- Parameters:
- Returns:
True if any result is truthy, False otherwise.
- Return type:
- map(*iterables, concurrency=None, task_name='', **kwargs)
Create a TaskMapping for this method.
- Parameters:
- Returns:
A TaskMapping instance for this method.
- Return type:
TaskMapping[I, T]
- async max(*iterables, concurrency=None, task_name='', **kwargs)
Find the maximum result.
- async min(*iterables, concurrency=None, task_name='', **kwargs)
Find the minimum result.
- async sum(*iterables, concurrency=None, task_name='', **kwargs)
Calculate the sum of the results.
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- _cache_handle: asyncio.TimerHandle
An asyncio handle used to pop the bound method from instance.__dict__ 5 minutes after its last use.
- _is_async_def
True if self.__wrapped__ is a coroutine function, False otherwise.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.method.ASyncBoundMethodSyncDefault[source]
Bases:
ASyncBoundMethod[I,P,T]A bound method with synchronous default behavior.
- __call__(*args, **kwargs)[source]
Call the bound method with synchronous default behavior.
- Parameters:
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(instance, unbound, async_def, **modifiers)
Initialize the bound method.
- Parameters:
instance (I) – The instance to bind the method to.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.
async_def (bool) – Whether the original function is an async def.
**modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- _should_await(kwargs)
Determine if the method should be awaited.
- async all(*iterables, concurrency=None, task_name='', **kwargs)
Check if all of the results are truthy.
- Parameters:
- Returns:
True if all results are truthy, False otherwise.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **kwargs)
Check if any of the results are truthy.
- Parameters:
- Returns:
True if any result is truthy, False otherwise.
- Return type:
- map(*iterables, concurrency=None, task_name='', **kwargs)
Create a TaskMapping for this method.
- Parameters:
- Returns:
A TaskMapping instance for this method.
- Return type:
TaskMapping[I, T]
- async max(*iterables, concurrency=None, task_name='', **kwargs)
Find the maximum result.
- async min(*iterables, concurrency=None, task_name='', **kwargs)
Find the minimum result.
- async sum(*iterables, concurrency=None, task_name='', **kwargs)
Calculate the sum of the results.
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- _cache_handle: asyncio.TimerHandle
An asyncio handle used to pop the bound method from instance.__dict__ 5 minutes after its last use.
- _is_async_def
True if self.__wrapped__ is a coroutine function, False otherwise.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.method.ASyncMethodDescriptor[source]
Bases:
ASyncDescriptor[I,P,T]This class provides the core functionality for creating
ASyncBoundMethodobjects, which can be used to define methods that can be called both synchronously and asynchronously.- async __call__(instance, *args, **kwargs)[source]
Asynchronously call the method.
- Parameters:
instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (ModifierKwargs) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _get_cache_handle(instance)[source]
Get a cache handle for the instance.
- Parameters:
instance (I) – The instance to create a cache handle for.
- Returns:
A timer handle for cache management.
- Return type:
TimerHandle
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(*instances, **bound_method_kwargs)
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.method.ASyncMethodDescriptorAsyncDefault[source]
Bases:
ASyncMethodDescriptor[I,P,T]A descriptor for asynchronous methods with an asynchronous default.
This class extends ASyncMethodDescriptor to provide an asynchronous default behavior for method calls.
- async __call__(instance, *args, **kwargs)
Asynchronously call the method.
- Parameters:
instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (ModifierKwargs) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _get_cache_handle(instance)
Get a cache handle for the instance.
- Parameters:
instance (I) – The instance to create a cache handle for.
- Returns:
A timer handle for cache management.
- Return type:
TimerHandle
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(*instances, **bound_method_kwargs)
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Asynchronous default version of the
all()method.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Asynchronous default version of the
any()method.
- default = 'async'
The default mode for this bound method. Always set to “async”.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Asynchronous default version of the
max()method.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Asynchronous default version of the
min()method.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Asynchronous default version of the
sum()method.
- wrapped
- class a_sync.a_sync.method.ASyncMethodDescriptorSyncDefault[source]
Bases:
ASyncMethodDescriptor[I,P,T]A descriptor for
ASyncBoundMethodSyncDefaultobjects.This class extends ASyncMethodDescriptor to provide a synchronous default behavior for method calls.
- async __call__(instance, *args, **kwargs)
Asynchronously call the method.
- Parameters:
instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (ModifierKwargs) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _get_cache_handle(instance)
Get a cache handle for the instance.
- Parameters:
instance (I) – The instance to create a cache handle for.
- Returns:
A timer handle for cache management.
- Return type:
TimerHandle
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(*instances, **bound_method_kwargs)
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Synchronous default version of the
all()method.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Synchronous default version of the
any()method.
- default = 'sync'
The default mode for this bound method. Always set to “sync”.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Synchronous default version of the
max()method.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Synchronous default version of the
min()method.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Synchronous default version of the
sum()method.
- wrapped
a_sync.a_sync.property module
- class a_sync.a_sync.property.ASyncCachedPropertyDescriptor[source]
Bases:
_ASyncPropertyDescriptorBase[I,T],AsyncCachedPropertyDescriptorA descriptor class for dual-function sync/async cached properties.
This class extends the API of ASyncPropertyDescriptor to provide caching functionality, storing the computed value after the first access.
- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)[source]
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.ASyncCachedPropertyDescriptorAsyncDefault[source]
Bases:
cached_property[I,T]A variant of
ASyncCachedPropertyDescriptorthat defaults to asynchronous behavior.This class is used for cached properties that are primarily intended to be accessed asynchronously but can also be used synchronously if needed.
- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.ASyncCachedPropertyDescriptorSyncDefault[source]
Bases:
cached_property[I,T]A variant of
ASyncCachedPropertyDescriptorthat defaults to synchronous behavior.This class is used for cached properties that are primarily intended to be accessed synchronously but can also be used asynchronously if needed.
- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.ASyncPropertyDescriptor[source]
Bases:
_ASyncPropertyDescriptorBase[I,T],AsyncPropertyDescriptor- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.ASyncPropertyDescriptorAsyncDefault[source]
-
A variant of
ASyncPropertyDescriptorthat defaults to asynchronous behavior.This class is used when the property is primarily intended to be accessed asynchronously but can also be used synchronously if needed.
- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- default = 'async'
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.ASyncPropertyDescriptorSyncDefault[source]
-
A variant of
ASyncPropertyDescriptorthat defaults to synchronous behavior.This class is used when the property is primarily intended to be accessed synchronously but can also be used asynchronously if needed.
- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- default = 'sync'
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.HiddenMethod[source]
Bases:
ASyncBoundMethodAsyncDefault[I,Tuple[()],T]- __call__(*args, **kwargs)
Call the bound method with asynchronous default behavior.
- __init__(instance, unbound, async_def, field_name, **modifiers)[source]
Initialize the bound method.
- Parameters:
instance (I) – The instance to bind the method to.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.
async_def (bool) – Whether the original function is an async def.
**modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.
field_name (str)
**modifiers
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **kwargs)
Check if all of the results are truthy.
- Parameters:
- Returns:
True if all results are truthy, False otherwise.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **kwargs)
Check if any of the results are truthy.
- Parameters:
- Returns:
True if any result is truthy, False otherwise.
- Return type:
- map(*iterables, concurrency=None, task_name='', **kwargs)
Create a TaskMapping for this method.
- Parameters:
- Returns:
A TaskMapping instance for this method.
- Return type:
TaskMapping[I, T]
- async max(*iterables, concurrency=None, task_name='', **kwargs)
Find the maximum result.
- async min(*iterables, concurrency=None, task_name='', **kwargs)
Find the minimum result.
- async sum(*iterables, concurrency=None, task_name='', **kwargs)
Calculate the sum of the results.
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- _cache_handle: asyncio.TimerHandle
An asyncio handle used to pop the bound method from instance.__dict__ 5 minutes after its last use.
- _is_async_def
True if self.__wrapped__ is a coroutine function, False otherwise.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.property.HiddenMethodDescriptor[source]
Bases:
ASyncMethodDescriptorAsyncDefault[I,Tuple[()],T]- async __call__(instance, *args, **kwargs)
Asynchronously call the method.
- Parameters:
instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(_fget, field_name=None, **modifiers)[source]
Initialize the HiddenMethodDescriptor.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[Awaitable[T]]] | Callable[[Concatenate[I, ~P]], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _get_cache_handle(instance)
Get a cache handle for the instance.
- Parameters:
instance (I) – The instance to create a cache handle for.
- Returns:
A timer handle for cache management.
- Return type:
TimerHandle
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(*instances, **bound_method_kwargs)
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- default = 'async'
The default mode for this bound method. Always set to “async”.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property._ASyncPropertyDescriptorBase[source]
Bases:
ASyncDescriptor[I,Tuple[()],T]- __init__(_fget, field_name=None, **modifiers)[source]
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(instances, owner=None, concurrency=None, name='')[source]
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.cached_property[source]
Bases:
ASyncCachedPropertyDescriptor[I,T]- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property.property[source]
Bases:
ASyncPropertyDescriptor[I,T]- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- a_sync.a_sync.property._parse_args(func, modifiers)[source]
- Parameters:
func (None | Literal['sync', 'async', None] | ~typing.Callable[[~a_sync._typing.I], ~typing.Awaitable[~a_sync._typing.T]])
modifiers (ModifierKwargs)
- Return type:
Tuple[Callable[[I], Awaitable[T]] | None, ModifierKwargs]
- a_sync.a_sync.property.a_sync_cached_property(func=None, **modifiers)[source]
- Parameters:
- Return type:
ASyncCachedPropertyDescriptor[I, T] | ASyncCachedPropertyDescriptorSyncDefault[I, T] | ASyncCachedPropertyDescriptorAsyncDefault[I, T] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], cached_property[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncCachedPropertyDescriptorSyncDefault[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncCachedPropertyDescriptorAsyncDefault[I, T]]
- a_sync.a_sync.property.a_sync_property(func=None, **modifiers)[source]
- Parameters:
- Return type:
ASyncPropertyDescriptor[I, T] | ASyncPropertyDescriptorSyncDefault[I, T] | ASyncPropertyDescriptorAsyncDefault[I, T] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], property[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncPropertyDescriptorSyncDefault[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncPropertyDescriptorAsyncDefault[I, T]]
a_sync.a_sync.singleton module
- class a_sync.a_sync.singleton.ASyncGenericSingleton[source]
Bases:
ASyncGenericBaseA base class for creating singleton-esque ASync classes.
This class combines the functionality of ASyncGenericBase with a singleton pattern, ensuring that only one instance of the class exists per execution mode (sync/async). It uses a custom metaclass to manage instance creation and caching.
Subclasses of ASyncGenericSingleton will have two instances instead of one: - one synchronous instance - one asynchronous instance
This allows for proper behavior in both synchronous and asynchronous contexts while maintaining the singleton pattern within each context.
Example
- class MyAsyncSingleton(ASyncGenericSingleton):
@a_sync def my_method(self):
# Method implementation
# These will return the same synchronous instance sync_instance1 = MyAsyncSingleton(sync=True) sync_instance2 = MyAsyncSingleton(sync=True)
# These will return the same asynchronous instance async_instance1 = MyAsyncSingleton(asynchronous=True) async_instance2 = MyAsyncSingleton(asynchronous=True)
assert sync_instance1 is sync_instance2 assert async_instance1 is async_instance2 assert sync_instance1 is not async_instance1
- __init__()
Module contents
- class a_sync.a_sync.ASyncCachedPropertyDescriptor[source]
Bases:
_ASyncPropertyDescriptorBase[I,T],AsyncCachedPropertyDescriptorA descriptor class for dual-function sync/async cached properties.
This class extends the API of ASyncPropertyDescriptor to provide caching functionality, storing the computed value after the first access.
- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)[source]
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.ASyncFunction[source]
Bases:
ModifiedMixin,Generic[P,T]A callable wrapper object that can be executed both synchronously and asynchronously.
This class wraps a function or coroutine function, allowing it to be called in both synchronous and asynchronous contexts. It provides a flexible interface for handling different execution modes and applying various modifiers to the function’s behavior.
The class supports various modifiers that can alter the behavior of the function, such as caching, rate limiting, and execution in specific contexts (e.g., thread pools).
Example
```python async def my_coroutine(x: int) -> str:
return str(x)
func = ASyncFunction(my_coroutine)
# Synchronous call result = func(5, sync=True) # returns “5”
# Asynchronous call result = await func(5) # returns “5” ```
- __call__(*args, **kwargs)[source]
Call the wrapped function either synchronously or asynchronously.
This method determines whether to execute the wrapped function synchronously or asynchronously based on the default mode and any provided flags.
- Parameters:
*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.
- Returns:
The result of the wrapped function call, which may be a coroutine if run asynchronously.
- Raises:
Exception – Any exception that may be raised by the wrapped function.
- Return type:
- __init__(fn, **modifiers)[source]
Initialize an ASyncFunction instance.
- Parameters:
fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to wrap.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)[source]
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Check if all results of the function applied to the iterables are truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Check if any result of the function applied to the iterables is truthy.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- map(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Create a TaskMapping for the wrapped function with the given iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[P, T]
- async max(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Find the maximum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The maximum result.
- Return type:
T
- async min(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Find the minimum result of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The minimum result.
- Return type:
T
- async sum(*iterables, concurrency=None, task_name='', **function_kwargs)[source]
Calculate the sum of the results of the function applied to the iterables.
- Parameters:
*iterables (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (int | None) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (~P) – Additional keyword arguments to pass to the function.
- Returns:
The sum of the results.
- Return type:
T
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.ASyncGenericBase[source]
Bases:
ASyncABCBase class for creating dual-function sync/async-capable classes without writing all your code twice.
This class 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.
- class a_sync.a_sync.ASyncPropertyDescriptor[source]
Bases:
_ASyncPropertyDescriptorBase[I,T],AsyncPropertyDescriptor- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.HiddenMethod[source]
Bases:
ASyncBoundMethodAsyncDefault[I,Tuple[()],T]- __call__(*args, **kwargs)
Call the bound method with asynchronous default behavior.
- __init__(instance, unbound, async_def, field_name, **modifiers)[source]
Initialize the bound method.
- Parameters:
instance (I) – The instance to bind the method to.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.
async_def (bool) – Whether the original function is an async def.
**modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.
field_name (str)
**modifiers
- Return type:
None
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _run_sync(kwargs)
Determine whether to run the function synchronously or asynchronously.
This method checks for a flag in the kwargs and defers to it if present. If no flag is specified, it defers to the default execution mode.
- async all(*iterables, concurrency=None, task_name='', **kwargs)
Check if all of the results are truthy.
- Parameters:
- Returns:
True if all results are truthy, False otherwise.
- Return type:
- async any(*iterables, concurrency=None, task_name='', **kwargs)
Check if any of the results are truthy.
- Parameters:
- Returns:
True if any result is truthy, False otherwise.
- Return type:
- map(*iterables, concurrency=None, task_name='', **kwargs)
Create a TaskMapping for this method.
- Parameters:
- Returns:
A TaskMapping instance for this method.
- Return type:
TaskMapping[I, T]
- async max(*iterables, concurrency=None, task_name='', **kwargs)
Find the maximum result.
- async min(*iterables, concurrency=None, task_name='', **kwargs)
Find the minimum result.
- async sum(*iterables, concurrency=None, task_name='', **kwargs)
Calculate the sum of the results.
- property _async_def: bool
Check if the wrapped function is an asynchronous function.
- Returns:
True if the wrapped function is an asynchronous function, False otherwise.
- property _async_wrap
The final wrapper if the wrapped function is an asynchronous function.
This method applies the appropriate modifiers and determines whether to await the result.
- Returns:
The final wrapped function.
- property _asyncified: Callable[[P], Awaitable[T]]
Convert the wrapped function to an asynchronous function and apply both sync and async modifiers.
- Returns:
An asynchronous function with both sync and async modifiers applied.
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- _cache_handle: asyncio.TimerHandle
An asyncio handle used to pop the bound method from instance.__dict__ 5 minutes after its last use.
- _is_async_def
True if self.__wrapped__ is a coroutine function, False otherwise.
- property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]
Apply modifiers to the wrapped function.
If the wrapped function is an asynchronous function, this method applies async modifiers. If the wrapped function is a synchronous function, this method applies sync modifiers.
- Returns:
The wrapped function with modifiers applied.
- property _sync_default: bool
Determine the default execution mode (sync or async) for the function.
If the user did not specify a default, this method defers to the function’s definition (sync vs async def).
- Returns:
True if the default is sync, False if the default is async.
- property _sync_wrap
The final wrapper if the wrapped function is a synchronous function.
This method applies the appropriate modifiers and determines whether to run the function synchronously or asynchronously.
- Returns:
The final wrapped function.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- property fn
Returns the final wrapped version of
ASyncFunction._fndecorated with all of the a_sync goodness.- Returns:
The final wrapped function.
- modifiers: ModifierManager
A
ModifierManagerinstance managing function modifiers.
- wrapped
- class a_sync.a_sync.HiddenMethodDescriptor[source]
Bases:
ASyncMethodDescriptorAsyncDefault[I,Tuple[()],T]- async __call__(instance, *args, **kwargs)
Asynchronously call the method.
- Parameters:
instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.
- Returns:
The result of the method call.
- Return type:
T
- __init__(_fget, field_name=None, **modifiers)[source]
Initialize the HiddenMethodDescriptor.
- Parameters:
_fget (Callable[[Concatenate[I, ~P]], Awaitable[Awaitable[T]]] | Callable[[Concatenate[I, ~P]], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _get_cache_handle(instance)
Get a cache handle for the instance.
- Parameters:
instance (I) – The instance to create a cache handle for.
- Returns:
A timer handle for cache management.
- Return type:
TimerHandle
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- map(*instances, **bound_method_kwargs)
Create a TaskMapping for the given instances.
- Parameters:
*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (~P) – Additional keyword arguments for the bound method.
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- default = 'async'
The default mode for this bound method. Always set to “async”.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.cached_property[source]
Bases:
ASyncCachedPropertyDescriptor[I,T]- __init__(_fget, _fset=None, _fdel=None, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- _check_method_name(method, method_type)
- _check_method_sync(method, method_type)
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- already_loaded(instance)
- del_cache_value(instance)
- deleter(method)
- get_cache(instance)
- get_cache_value(instance)
- get_instance_state(instance)
- has_cache_value(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- not_loaded(instance)
- set_cache_value(instance, value)
- setter(method)
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- class a_sync.a_sync.property[source]
Bases:
ASyncPropertyDescriptor[I,T]- __init__(_fget, field_name=None, **modifiers)
Initialize the {cls}.
- Parameters:
_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
field_name (str | None) – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.
- Raises:
ValueError – If _fget is not callable.
- Return type:
None
- async _all(*instances, concurrency=None, name='', **kwargs)
Check if all results are truthy.
- Parameters:
- Returns:
A boolean indicating if all results are truthy.
- Return type:
- async _any(*instances, concurrency=None, name='', **kwargs)
Check if any result is truthy.
- Parameters:
- Returns:
A boolean indicating if any result is truthy.
- Return type:
- _asyncify(func)
Convert a synchronous function to an asynchronous one and apply async modifiers.
- async _max(*instances, concurrency=None, name='', **kwargs)
Find the maximum result.
- async _min(*instances, concurrency=None, name='', **kwargs)
Find the minimum result.
- async _sum(*instances, concurrency=None, name='', **kwargs)
Calculate the sum of results.
- awaitable_only(instance)
- get_loader(instance)
- map(instances, owner=None, concurrency=None, name='')
Create a TaskMapping for the given instances.
- Parameters:
- Returns:
A TaskMapping object.
- Return type:
TaskMapping[I, T]
- property _await: Callable[[Awaitable[T]], T]
Apply sync modifiers to the _helpers._await function and cache it.
- Returns:
A function that applies sync modifiers to awaitable objects.
- property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if all results are truthy.- Returns:
An ASyncFunction object.
- property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], bool]
Create an
ASyncFunctionthat checks if any result is truthy.- Returns:
An ASyncFunction object.
- property default: Literal['sync', 'async', None]
Get the default execution mode (sync, async, or None) for the function.
- Returns:
The default execution mode as determined by the modifiers.
- field_name
The name of the field the {cls} is bound to.
- property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the maximum result.- Returns:
An ASyncFunction object.
- property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the minimum result.- Returns:
An ASyncFunction object.
- modifiers: ModifierManager
- property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[I], P], T]
Create an
ASyncFunctionthat returns the sum of results.- Returns:
An ASyncFunction object.
- wrapped
- a_sync.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
ModifierKwargsfor available options.
- Return type:
ASyncDecorator | ASyncFunction[~P, T]
- Modifiers:
- lib defaults:
- async settings
cache_type: CacheType = None, - This can be None or ‘memory’. ‘memory’ is a lru cache which can be modified with the ‘cache_typed’,’ram_cache_maxsize’,’ram_cache_ttl’ modifiers. cache_typed: bool = False, - Set to True if you want types considered treated for cache keys. ie with cache_typed=True, Decimal(0) and 0 will be considered separate keys. ram_cache_maxsize: Optional[int] = -1, - The maxsize for your lru cache. None if cache is unbounded. If you set this value without specifying a cache type, ‘memory’ will automatically be applied. ram_cache_ttl: Optional[int] = None, - 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: Optional[int] = None, - Setting this value enables a rate limiter for the decorated function. semaphore: SemaphoreSpec = None, - drop in a Semaphore for your async defined functions.
- sync settings
executor: Executor = config.default_sync_executor
- Returns:
An
ASyncDecoratorif used as a decorator factory, or anASyncFunctionif used directly on a function.- Parameters:
coro_fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T] | None)
default (Literal['sync', 'async', None] | None)
modifiers (Unpack[ModifierKwargs])
- Return type:
ASyncDecorator | ASyncFunction[~P, T]
Examples
The decorator can be used in several ways.
1. As a simple decorator:
`python >>> @a_sync ... async def some_async_fn(): ... return True >>> await some_fn() True >>> some_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 0x12345678> `2. As a decorator with default mode specified:
`python >>> @a_sync(default='sync') ... async def some_fn(): ... return True ... >>> some_fn() True `3. As a decorator with modifiers:
`python >>> @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:
`python >>> 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:
`python result = some_fn() # Synchronous call result = await some_fn() # Asynchronous call `The execution mode can also be explicitly specified during the call:
`python 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.