a_sync.a_sync package 

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x * 2

instance = MyClass() result = instance.my_method.map([1, 2, 3])

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

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 or function calls will be synchronous or asynchronous.

You can use any of the provided flags, whichever makes the most sense for your use case.

AFFIRMATIVE_FLAGS: Set of flags indicating synchronous behavior. Currently includes “sync”.

NEGATIVE_FLAGS: Set of flags indicating asynchronous behavior. Currently includes “asynchronous”.

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.get_event_loop()

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(dict kwargs) → str

Return type:: str

a_sync.a_sync._meta module

class a_sync.a_sync._meta.ASyncMeta[source]

Bases: ABCMeta

Metaclass for wrapping class attributes with asynchronous capabilities.

Any class with ASyncMeta as its metaclass will have its functions and properties wrapped with asynchronous capabilities upon class instantiation. This includes wrapping functions with ASyncMethodDescriptor and properties with ASyncPropertyDescriptor or ASyncCachedPropertyDescriptor. It also handles attributes that are instances of ASyncFunction, which are used when functions are decorated with a_sync decorators to apply specific modifiers to those functions.

Attributes that are instances of _ASyncFutureWrappedFn and Semaphore are explicitly skipped and not wrapped.

Example

To create a class with asynchronous capabilities, define your class with ASyncMeta as its metaclass:

>>> class MyClass(metaclass=ASyncMeta):
...     def my_method(self):
...         return "Hello, World!"

The my_method function will be wrapped with ASyncMethodDescriptor, allowing it to be used asynchronously.

See also

ASyncFunction
ASyncMethodDescriptor
ASyncPropertyDescriptor
ASyncCachedPropertyDescriptor

__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

Metaclass for creating singleton instances with asynchronous capabilities.

This metaclass extends ASyncMeta to ensure that only one instance of a class is created for each synchronous or asynchronous context.

Example

To create a singleton class with asynchronous capabilities, define your class with ASyncSingletonMeta as its metaclass:

>>> class MySingleton(metaclass=ASyncSingletonMeta):
...     def __init__(self):
...         print("Instance created")

The MySingleton class will ensure that only one instance is created for each context.

See also

ASyncMeta

__call__(*args, **kwargs)[source]

Call self as a function.

Parameters:

args (Any)
kwargs (Any)

__init__(name, bases, namespace)[source]

Parameters:

name (str)
bases (Tuple[type, ...])
namespace (Dict[str, Any])

Return type:

None

_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

This module provides an abstract base class for defining asynchronous and synchronous behavior.

The ASyncABC class uses the ASyncMeta metaclass to facilitate the creation of classes that can operate in both asynchronous and synchronous contexts. It provides concrete methods to determine the execution mode based on flags and keyword arguments.

Note: It is recommended to use ASyncGenericBase for most use cases. This class is intended for more custom implementations if necessary.

class a_sync.a_sync.abstract.ASyncABC

Bases: object

Abstract Base Class for defining asynchronous and synchronous behavior.

This class provides methods to determine the execution mode based on flags and keyword arguments. It is designed to be subclassed, allowing developers to create classes that can be used in both synchronous and asynchronous contexts.

See also

ASyncGenericBase: A more user-friendly base class for creating dual-mode classes.
ASyncMeta: Metaclass that facilitates asynchronous capabilities in class attributes.

Examples

To create a class that inherits from ASyncABC, you need to implement the abstract methods and can override the concrete methods if needed.

```python class MyASyncClass(ASyncABC):

```

In this example, MyASyncClass is a subclass of ASyncABC with custom implementations for the required abstract methods.

__init__(self) → None

Return type:: None

a_sync.a_sync.base module

class a_sync.a_sync.base.ASyncGenericBase

Bases: ASyncABC

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

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

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

Example

```python class MyClass(ASyncGenericBase):

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

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

Note

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

__init__(self)

a_sync.a_sync.config module

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.

Environment Variables:

A_SYNC_EXECUTOR_TYPE: Specifies the type of executor to use. Valid values are: strings that start with ‘p’ for ProcessPoolExecutor (e.g., ‘processes’) or ‘t’ for ThreadPoolExecutor (e.g., ‘threads’). Defaults to ‘threads’.
A_SYNC_EXECUTOR_VALUE: Specifies the number of workers for the executor.: Defaults to 8.

A_SYNC_DEFAULT_MODE: Sets the default mode for a_sync functions if not specified. A_SYNC_CACHE_TYPE: Sets the default cache type. If not specified, defaults to None. A_SYNC_CACHE_TYPED: Boolean flag to determine if cache keys should consider types. A_SYNC_RAM_CACHE_MAXSIZE: Sets the maximum size for the RAM cache. Defaults to -1. A_SYNC_RAM_CACHE_TTL: Sets the time-to-live for cache entries. If not specified,

A_SYNC_RUNS_PER_MINUTE: Sets the rate limit for function execution. A_SYNC_SEMAPHORE: Sets the semaphore limit for function execution.

Examples

To set the executor type to use threads with 4 workers, set the environment variables:

export A_SYNC_EXECUTOR_TYPE=threads
export A_SYNC_EXECUTOR_VALUE=4

To configure caching with a maximum size of 100 and a TTL of 60 seconds:

export A_SYNC_CACHE_TYPE=memory
export A_SYNC_RAM_CACHE_MAXSIZE=100
export A_SYNC_RAM_CACHE_TTL=60

TODO: explain how and where these values are used

See also

concurrent.futures: For more details on executors.
functools: For caching mechanisms.

a_sync.a_sync.config.get_default_executor()[source]

Get the default executor based on the EXECUTOR_TYPE environment variable.

Returns:: An instance of either ProcessPoolExecutor or ThreadPoolExecutor.
Raises:: ValueError – If an invalid EXECUTOR_TYPE is specified. Valid values are
Return type:: Executor

:raises strings that start with ‘p’ for ProcessPoolExecutor: :raises or ‘t’ for ThreadPoolExecutor.:

Examples

>>> import os
>>> os.environ["A_SYNC_EXECUTOR_TYPE"] = "threads"
>>> executor = get_default_executor()
>>> isinstance(executor, ThreadPoolExecutor)
True

>>> os.environ["A_SYNC_EXECUTOR_TYPE"] = "processes"
>>> executor = get_default_executor()
>>> isinstance(executor, ProcessPoolExecutor)
True

a_sync.a_sync.config.CACHE_TYPE: Literal['memory', None] = None: Sets the default cache type. If not specified, defaults to None.

a_sync.a_sync.config.CACHE_TYPED = False: Boolean flag to determine if cache keys should consider types.

a_sync.a_sync.config.DEFAULT_MODE: Literal['sync', 'async', None] = None: Sets the default mode for a_sync functions if not specified.

a_sync.a_sync.config.EXECUTOR_TYPE = 'threads'

Specifies the type of executor to use.

Valid values are strings that start with ‘p’ for ProcessPoolExecutor (e.g., ‘processes’) or ‘t’ for ThreadPoolExecutor (e.g., ‘threads’). Defaults to ‘threads’.

a_sync.a_sync.config.EXECUTOR_VALUE = 8: Specifies the number of workers for the executor. Defaults to 8.

a_sync.a_sync.config.RAM_CACHE_MAXSIZE = -1: Sets the maximum size for the RAM cache. Defaults to -1. # TODO: explain what -1 does

a_sync.a_sync.config.RAM_CACHE_TTL = None: Sets the time-to-live for cache entries. If not specified, defaults to 0, which is then checked against null_modifiers[“ram_cache_ttl”] to potentially set it to None, meaning cache entries do not expire by default.

a_sync.a_sync.config.RUNS_PER_MINUTE = None: Sets the rate limit for function execution.

a_sync.a_sync.config.SEMAPHORE = None: Sets the semaphore limit for function execution.

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 ModifierKwargs for available options.

Return type:

ASyncDecorator | ASyncFunction[~P, T]

Modifiers:

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

cache_type: Can be None or ‘memory’. ‘memory’ is an LRU cache which can be modified with the ‘cache_typed’, ‘ram_cache_maxsize’, and ‘ram_cache_ttl’ modifiers.
cache_typed: Set to True if you want types considered for cache keys. For example, with cache_typed=True, Decimal(0) and 0 will be considered separate keys.
ram_cache_maxsize: The max size for your LRU cache. None if the cache is unbounded. If you set this value without specifying a cache type, ‘memory’ will automatically be applied.
ram_cache_ttl: The TTL for items in your LRU cache. Set to None. If you set this value without specifying a cache type, ‘memory’ will automatically be applied.
runs_per_minute: Setting this value enables a rate limiter for the decorated function.
semaphore: Drop in a Semaphore for your async defined functions.
executor: The executor for the synchronous function. Set to the library’s default of config.default_sync_executor.

Examples

The decorator can be used in several ways.

As a simple decorator:

>>> @a_sync
... async def some_async_fn():
...     return True
>>> await some_async_fn()
True
>>> some_async_fn(sync=True)
True

>>> @a_sync
... def some_sync_fn():
...     return True
>>> some_sync_fn()
True
>>> some_sync_fn(sync=False)
<coroutine object some_sync_fn at 0x7fb4f5fb49c0>

As a decorator with default mode specified:

>>> @a_sync(default='sync')
... async def some_fn():
...     return True
...
>>> some_fn()
True
>>> some_fn(sync=False)
<coroutine object some_fn at 0x7fb4f5fb49c0>

>>> @a_sync('async')
... def some_fn():
...     return True
...
>>> some_fn()
<coroutine object some_fn at 0x7fb4f5fb49c0>
>>> some_fn(asynchronous=False)
True

As a decorator with modifiers:

>>> @a_sync(cache_type='memory', runs_per_minute=60)
... async def some_fn():
...    return True
...
>>> some_fn(sync=True)
True

Applied directly to a function:

>>> some_fn = a_sync(some_existing_function, default='sync')
>>> some_fn()
"some return value"

The decorated function can then be called either synchronously or asynchronously:

>>> result = some_fn()  # Synchronous call
>>> result = await some_fn()  # Asynchronous call

The execution mode can also be explicitly specified during the call:

>>> result = some_fn(sync=True)  # Force synchronous execution
>>> result = await some_fn(sync=False)  # Force asynchronous execution

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

Note

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

See also

ASyncFunction, ASyncDecorator

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 or function calls will be synchronous or asynchronous.

You can use any of the provided flags, whichever makes the most sense for your use case.

AFFIRMATIVE_FLAGS: Set of flags indicating synchronous behavior. Currently includes “sync”.

NEGATIVE_FLAGS: Set of flags indicating asynchronous behavior. Currently includes “asynchronous”.

VIABLE_FLAGS: Set of all valid flags, combining both synchronous and asynchronous indicators.

a_sync.a_sync.function module

class a_sync.a_sync.function.ASyncDecorator

Bases: _ModifiedMixin

__call__(**kwds): Helper for @overload to raise when called.

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

modifiers

class a_sync.a_sync.function.ASyncDecoratorAsyncDefault

Bases: ASyncDecorator

__call__(**kwds): Helper for @overload to raise when called.

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

modifiers

class a_sync.a_sync.function.ASyncDecoratorSyncDefault

Bases: ASyncDecorator

__call__(**kwds): Helper for @overload to raise when called.

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

modifiers

class a_sync.a_sync.function.ASyncFunction

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

Note

The logic for determining whether to execute the function synchronously or asynchronously is handled by the self.fn property, which checks for flags in the kwargs and defers to the default execution mode if no flags are specified.

Example

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”

See also

_ModifiedMixin
ModifierManager

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

Calls 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. The decision logic is encapsulated within the self.fn property, which uses the _run_sync method to decide the execution mode.

Note

The self.fn property is responsible for selecting the appropriate execution path (sync or async) by leveraging the _run_sync method.

Parameters:

*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Return type:

Coroutine[Any, Any, T] | T

See also

default
_run_sync()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.function.ASyncFunctionAsyncDefault

Bases: ASyncFunction[P, T]

A specialized ASyncFunction that 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

@a_sync(default=’async’) async def my_function(x: int) -> str:

# Asynchronous call (default behavior) result = await my_function(5) # returns “5”

# Synchronous call result = my_function(5, sync=True) # returns “5”

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

Calls 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.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Returns:

The result of the function call.

Return type:

Coroutine[Any, Any, T] | T

See also

ASyncFunction.__call__()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.function.ASyncFunctionSyncDefault

Bases: ASyncFunction[P, T]

A specialized ASyncFunction that 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

@a_sync(default=’sync’) async def my_function(x: int) -> str:

# Synchronous call (default behavior) result = my_function(5) # returns “5”

# Asynchronous call result = await my_function(5, sync=False) # returns “5”

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

Calls 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.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Returns:

The result of the function call.

Return type:

Coroutine[Any, Any, T] | T

See also

ASyncFunction.__call__()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.function._ModifiedMixin

Bases: object

A mixin class for internal use that provides functionality for applying modifiers to functions.

This class is used as a base for ASyncFunction and its variants, such as ASyncFunctionAsyncDefault and ASyncFunctionSyncDefault, to handle the application of async and sync modifiers to functions. Modifiers can alter the behavior of functions, such as converting sync functions to async, applying caching, or rate limiting.

See also

ASyncFunction
ModifierManager

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

modifiers

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

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. It handles caching of bound methods and includes logic for determining whether to await the method call based on flags or default settings.

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptor
...     async def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> await obj.my_method()
42
>>> obj.my_method(sync=True)
42

See also

ASyncMethodDescriptor
ASyncFunction

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

Call the bound method.

This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.

Parameters:

*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

Coroutine[Any, Any, T] | T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> await bound_method(arg1, arg2, kwarg1=value1)
>>> bound_method(arg1, arg2, kwarg1=value1, sync=True)

__init__(_ModifiedMixin self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:

instance (I) – The instance to bind the method to.
unbound (AnyFn[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

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptor
...     async def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> bound_method = ASyncBoundMethod(obj, MyClass.my_method, True)

async all(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if all of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.all(iterable1, iterable2)

async any(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if any of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.any(iterable1, iterable2)

map(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for this method.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Returns:

A TaskMapping instance for this method.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> task_mapping = bound_method.map(iterable1, iterable2, concurrency=5)
TODO briefly include how someone would then use task_mapping

async max(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.max(iterable1, iterable2)

async min(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.min(iterable1, iterable2)

async sum(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of the results.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.sum(iterable1, iterable2)

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_cache_handle: TimerHandle = None

_fn = None

_is_async_def

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.method.ASyncBoundMethodAsyncDefault

Bases: ASyncBoundMethod[I, P, T]

A bound method with asynchronous default behavior.

This class is a specialized version of ASyncBoundMethod that defaults to asynchronous execution. It overrides the __call__ method to enforce asynchronous default behavior.

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptorAsyncDefault
...     async def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> await obj.my_method()
42
>>> obj.my_method(sync=True)
42

See also

ASyncBoundMethod
ASyncMethodDescriptorAsyncDefault

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

Call the bound method.

This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.

Parameters:

*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

Coroutine[Any, Any, T] | T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> await bound_method(arg1, arg2, kwarg1=value1)
>>> bound_method(arg1, arg2, kwarg1=value1, sync=True)

__init__(_ModifiedMixin self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:

instance (I) – The instance to bind the method to.
unbound (AnyFn[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

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptor
...     async def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> bound_method = ASyncBoundMethod(obj, MyClass.my_method, True)

async all(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if all of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.all(iterable1, iterable2)

async any(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if any of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.any(iterable1, iterable2)

map(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for this method.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Returns:

A TaskMapping instance for this method.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> task_mapping = bound_method.map(iterable1, iterable2, concurrency=5)
TODO briefly include how someone would then use task_mapping

async max(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.max(iterable1, iterable2)

async min(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.min(iterable1, iterable2)

async sum(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of the results.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.sum(iterable1, iterable2)

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_cache_handle = None

_fn = None

_is_async_def

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.method.ASyncBoundMethodSyncDefault

Bases: ASyncBoundMethod[I, P, T]

A bound method with synchronous default behavior.

This class is a specialized version of ASyncBoundMethod that defaults to synchronous execution. It overrides the __call__ method to enforce synchronous default behavior.

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptorSyncDefault
...     def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> obj.my_method()
42
>>> await obj.my_method(sync=False)
42

See also

ASyncBoundMethod
ASyncMethodDescriptorSyncDefault

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

Call the bound method.

This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.

Parameters:

*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

Coroutine[Any, Any, T] | T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> await bound_method(arg1, arg2, kwarg1=value1)
>>> bound_method(arg1, arg2, kwarg1=value1, sync=True)

__init__(_ModifiedMixin self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:

instance (I) – The instance to bind the method to.
unbound (AnyFn[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

Examples

>>> class MyClass:
...     def __init__(self, value):
...         self.value = value
...
...     @ASyncMethodDescriptor
...     async def my_method(self):
...         return self.value
...
>>> obj = MyClass(42)
>>> bound_method = ASyncBoundMethod(obj, MyClass.my_method, True)

async all(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if all of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.all(iterable1, iterable2)

async any(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if any of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.any(iterable1, iterable2)

map(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for this method.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Returns:

A TaskMapping instance for this method.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> task_mapping = bound_method.map(iterable1, iterable2, concurrency=5)
TODO briefly include how someone would then use task_mapping

async max(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.max(iterable1, iterable2)

async min(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.min(iterable1, iterable2)

async sum(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of the results.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.sum(iterable1, iterable2)

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_cache_handle = None

_fn = None

_is_async_def

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.method.ASyncMethodDescriptor

Bases: ASyncDescriptor[I, P, T]

A descriptor for managing methods that can be called both synchronously and asynchronously.

This class provides the core functionality for binding methods to instances and determining the execution mode (“sync” or “async”) based on various conditions, such as the instance type, the method’s default setting, or specific flags passed during the method call.

The descriptor is responsible for creating an appropriate bound method when accessed, which is the actual object that can be called in both synchronous and asynchronous contexts. It can create different types of bound methods (ASyncBoundMethodSyncDefault, ASyncBoundMethodAsyncDefault, or ASyncBoundMethod) based on the default mode or instance type.

If the default mode is explicitly set to “sync” or “async”, it creates ASyncBoundMethodSyncDefault or ASyncBoundMethodAsyncDefault respectively. If neither is set, it defaults to creating an ASyncBoundMethod. For instances of ASyncABC, it checks the __a_sync_instance_should_await__ attribute to decide the type of bound method to create.

It also manages cache handles for bound methods and prevents setting or deleting the descriptor.

Examples

>>> class MyClass:
...     @ASyncMethodDescriptor
...     async def my_method(self):
...         return "Hello, World!"
...
>>> obj = MyClass()
>>> await obj.my_method()
'Hello, World!'
>>> obj.my_method(sync=True)
'Hello, World!'

See also

ASyncBoundMethod
ASyncFunction

async __call__(self, instance: I, *args: P.args, **kwargs: P.kwargs) → T

Asynchronously call the method.

Parameters:

instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

T

Examples

>>> descriptor = ASyncMethodDescriptor(my_async_function)
>>> await descriptor(instance, arg1, arg2, kwarg1=value1)

__init__(_ModifiedMixin self, _fget: AnyFn[Concatenate[I, P], T], field_name: Optional[str] = None, **modifiers: ModifierKwargs) → None

Initialize the ASyncDescriptor.

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(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x * 2

instance = MyClass() result = instance.my_method.map([1, 2, 3])

_initialized = False

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.method.ASyncMethodDescriptorAsyncDefault

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. It specifically creates ASyncBoundMethodAsyncDefault instances, which are specialized versions of ASyncBoundMethod with asynchronous default behavior.

Examples

>>> class MyClass:
...     @ASyncMethodDescriptorAsyncDefault
...     async def my_method(self):
...         return "Hello, World!"
...
>>> obj = MyClass()
>>> coro = obj.my_method()
>>> coro
<coroutine object MyClass.my_method at 0x7fb4f5fb49c0>
>>> await coro
>>> obj.my_method(sync=True)
'Hello, World!'

See also

ASyncBoundMethodAsyncDefault
ASyncFunctionAsyncDefault

async __call__(self, instance: I, *args: P.args, **kwargs: P.kwargs) → T

Asynchronously call the method.

Parameters:

instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

T

Examples

>>> descriptor = ASyncMethodDescriptor(my_async_function)
>>> await descriptor(instance, arg1, arg2, kwarg1=value1)

__init__(_ModifiedMixin self, _fget: AnyFn[Concatenate[I, P], T], field_name: Optional[str] = None, **modifiers: ModifierKwargs) → None

Initialize the ASyncDescriptor.

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(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x * 2

instance = MyClass() result = instance.my_method.map([1, 2, 3])

_initialized = False

all: ASyncFunctionAsyncDefault[Concatenate[AnyIterable[I], P], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionAsyncDefault[Concatenate[AnyIterable[I], P], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'async'

field_name

max: ASyncFunctionAsyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionAsyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionAsyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.method.ASyncMethodDescriptorSyncDefault

Bases: ASyncMethodDescriptor[I, P, T]

A descriptor for ASyncBoundMethodSyncDefault objects.

This class extends ASyncMethodDescriptor to provide a synchronous default behavior for method calls. It specifically creates ASyncBoundMethodSyncDefault instances, which are specialized versions of ASyncBoundMethod with synchronous default behavior.

Examples

>>> class MyClass:
...     @ASyncMethodDescriptorSyncDefault
...     def my_method(self):
...         return "Hello, World!"
...
>>> obj = MyClass()
>>> obj.my_method()
'Hello, World!'
>>> coro = obj.my_method(sync=False)
>>> coro
<coroutine object MyClass.my_method at 0x7fb4f5fb49c0>
>>> await coro
'Hello, World!'

See also

ASyncBoundMethodSyncDefault
ASyncFunctionSyncDefault

async __call__(self, instance: I, *args: P.args, **kwargs: P.kwargs) → T

Asynchronously call the method.

Parameters:

instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

T

Examples

>>> descriptor = ASyncMethodDescriptor(my_async_function)
>>> await descriptor(instance, arg1, arg2, kwarg1=value1)

__init__(_ModifiedMixin self, _fget: AnyFn[Concatenate[I, P], T], field_name: Optional[str] = None, **modifiers: ModifierKwargs) → None

Initialize the ASyncDescriptor.

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(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x * 2

instance = MyClass() result = instance.my_method.map([1, 2, 3])

_initialized = False

all: ASyncFunctionSyncDefault[Concatenate[AnyIterable[I], P], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionSyncDefault[Concatenate[AnyIterable[I], P], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'sync'

field_name

max: ASyncFunctionSyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionSyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionSyncDefault[Concatenate[AnyIterable[I], P], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

a_sync.a_sync.property module

class a_sync.a_sync.property.ASyncCachedPropertyDescriptor

Bases: _ASyncPropertyDescriptorBase[I, T], AsyncCachedPropertyDescriptor

A 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__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.ASyncCachedPropertyDescriptorAsyncDefault

A variant of ASyncCachedPropertyDescriptor that 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.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

__init__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default: Literal['async']

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.ASyncCachedPropertyDescriptorSyncDefault

A variant of ASyncCachedPropertyDescriptor that 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.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

__init__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default: Literal['sync']

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.ASyncPropertyDescriptor

Bases: _ASyncPropertyDescriptorBase[I, T], AsyncPropertyDescriptor

Descriptor class for asynchronous properties.

__init__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.ASyncPropertyDescriptorAsyncDefault

Bases: ASyncPropertyDescriptor[I, T]

A variant of ASyncPropertyDescriptor that 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__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all: ASyncFunctionAsyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionAsyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'async'

field_name

hidden_method_descriptor

hidden_method_name

max: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.ASyncPropertyDescriptorSyncDefault

Bases: ASyncPropertyDescriptor[I, T]

A variant of ASyncPropertyDescriptor that 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__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all: ASyncFunctionSyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionSyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'sync'

field_name

hidden_method_descriptor

hidden_method_name

max: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property.HiddenMethod

Bases: ASyncBoundMethodAsyncDefault[I, Tuple[()], T]

Represents a hidden method for asynchronous properties.

This class is used internally to manage hidden getter methods associated with a/sync properties.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

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

Call the bound method.

This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.

Parameters:

*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

Coroutine[Any, Any, T] | T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> await bound_method(arg1, arg2, kwarg1=value1)
>>> bound_method(arg1, arg2, kwarg1=value1, sync=True)

__init__(self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, unicode field_name: str, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the HiddenMethod.

Parameters:

instance (I) – The instance to which the method is bound.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function to be wrapped.
async_def (bool) – Indicates if the method is asynchronous.
field_name (str) – The name of the field associated with the method.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async all(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if all of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.all(iterable1, iterable2)

async any(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if any of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.any(iterable1, iterable2)

map(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for this method.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Returns:

A TaskMapping instance for this method.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> task_mapping = bound_method.map(iterable1, iterable2, concurrency=5)
TODO briefly include how someone would then use task_mapping

async max(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.max(iterable1, iterable2)

async min(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.min(iterable1, iterable2)

async sum(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of the results.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.sum(iterable1, iterable2)

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_cache_handle = None

_fn = None

_is_async_def

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.property.HiddenMethodDescriptor

Bases: ASyncMethodDescriptorAsyncDefault[I, Tuple[()], T]

Descriptor for hidden methods associated with asynchronous properties.

This class is used internally to manage hidden getter methods associated with a/sync properties.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

async __call__(self, instance: I, *args: P.args, **kwargs: P.kwargs) → T

Asynchronously call the method.

Parameters:

instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

T

Examples

>>> descriptor = ASyncMethodDescriptor(my_async_function)
>>> await descriptor(instance, arg1, arg2, kwarg1=value1)

__init__(self, _fget: AnyFn[Concatenate[I, P], Awaitable[T]], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

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(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x * 2

instance = MyClass() result = instance.my_method.map([1, 2, 3])

_initialized = False

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'async'

field_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.property._ASyncPropertyDescriptorBase

Bases: ASyncDescriptor[I, Tuple[()], T]

Base class for creating asynchronous properties.

This class provides the foundation for defining properties that can be accessed both synchronously and asynchronously. It includes utility methods for common operations such as any, all, min, max, and sum.

__init__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping: Type[TaskMapping] = None

all: ASyncFunction[AnyIterable[I], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunction[AnyIterable[I], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor: HiddenMethodDescriptor[T]

hidden_method_name

max: ASyncFunction[AnyIterable[I], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunction[AnyIterable[I], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunction[AnyIterable[I], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

a_sync.a_sync.property.a_sync_cached_property(func: AnyGetterFunction[I, T] | None = None, **modifiers: Unpack[ModifierKwargs]) → ASyncCachedPropertyDescriptor[I, T] | ASyncCachedPropertyDescriptorSyncDefault[I, T] | ASyncCachedPropertyDescriptorAsyncDefault[I, T] | ASyncCachedPropertyDecorator[I, T] | ASyncCachedPropertyDecoratorSyncDefault[I, T] | ASyncCachedPropertyDecoratorAsyncDefault[I, T]

Decorator for creating cached properties that can be accessed both synchronously and asynchronously.

Parameters:

func (Callable[[I], Awaitable[T]] | Callable[[I], T] | None) – The function to be wrapped.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Returns:

A cached property descriptor that supports both sync and async access.

Return type:

ASyncCachedPropertyDescriptor[I, T] | ASyncCachedPropertyDescriptorSyncDefault[I, T] | ASyncCachedPropertyDescriptorAsyncDefault[I, T] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncCachedPropertyDescriptor[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: AnyGetterFunction[I, T] | DefaultMode = None, **modifiers: Unpack[ModifierKwargs]) → ASyncPropertyDescriptor[I, T] | ASyncPropertyDescriptorSyncDefault[I, T] | ASyncPropertyDescriptorAsyncDefault[I, T] | ASyncPropertyDecorator[I, T] | ASyncPropertyDecoratorSyncDefault[I, T] | ASyncPropertyDecoratorAsyncDefault[I, T]

Decorator for creating properties that can be accessed both synchronously and asynchronously.

Parameters:

func (Callable[[I], Awaitable[T]] | Callable[[I], T] | Literal['sync', 'async', None] | None) – The function to be wrapped.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Returns:

A property descriptor that supports both sync and async access.

Return type:

ASyncPropertyDescriptor[I, T] | ASyncPropertyDescriptorSyncDefault[I, T] | ASyncPropertyDescriptorAsyncDefault[I, T] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncPropertyDescriptor[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: ASyncGenericBase

A 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 ASyncSingletonMeta 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.

Note

This class can be instantiated directly, but it is intended to be subclassed to define specific asynchronous behavior. Subclasses should define the necessary properties and methods to specify the asynchronous behavior, as outlined in ASyncGenericBase.

Example

Create a subclass of ASyncGenericSingleton to define specific behavior:

class MyAsyncSingleton(ASyncGenericSingleton):
    @property
    def __a_sync_flag_name__(self):
        return "asynchronous"

    @property
    def __a_sync_flag_value__(self):
        return self.asynchronous

    @classmethod
    def __a_sync_default_mode__(cls):
        return False

    @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

See also

ASyncGenericBase for base functionality.
ASyncSingletonMeta for the metaclass managing the singleton behavior.

__init__(self)

Module contents

This module enables developers to write both synchronous and asynchronous code without having to write redundant code.

The two main objects you should use are

a decorator @a_sync()
a base class ASyncGenericBase which can be used to create classes that can be utilized in both synchronous and asynchronous contexts.

The rest of the objects are exposed for type checking only, you should not make use of them otherwise.

class a_sync.a_sync.ASyncCachedPropertyDescriptor

Bases: _ASyncPropertyDescriptorBase[I, T], AsyncCachedPropertyDescriptor

A 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__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.ASyncCachedPropertyDescriptorAsyncDefault

A variant of ASyncCachedPropertyDescriptor that 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.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

__init__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default: Literal['async']

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.ASyncCachedPropertyDescriptorSyncDefault

A variant of ASyncCachedPropertyDescriptor that 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.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

__init__(self, _fget: AsyncGetterFunction[I, T], _fset=None, _fdel=None, field_name=None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the ASyncCachedPropertyDescriptor.

Parameters:

_fget (Callable[[I], Awaitable[T]]) – The function to be wrapped.
_fset – Optional setter function for the property.
_fdel – Optional deleter function for the property.
field_name – Optional name for the field. If not provided, the function’s name will be used.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

_check_method_name(self, method, method_type)

_check_method_sync(self, method, method_type)

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

del_cache_value(self, instance)

deleter(self, method)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_cache(self, instance)

get_cache_value(self, instance)

get_instance_state(self, instance)

get_loader(self, instance: I) → Callable[[], T]

Retrieves the loader function for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: A callable that loads the property value.
Return type:: Callable[[], T]

get_lock(self, instance: I) → 'Task[T]'

Retrieves the lock for the property.

Parameters:: instance (I) – The instance from which the property is accessed.
Returns:: An asyncio Task representing the lock.
Return type:: Task[T]

has_cache_value(self, instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

set_cache_value(self, instance, value)

setter(self, method)

_TaskMapping = None

_load_value = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default: Literal['sync']

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.ASyncFunction

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

Note

The logic for determining whether to execute the function synchronously or asynchronously is handled by the self.fn property, which checks for flags in the kwargs and defers to the default execution mode if no flags are specified.

Example

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”

See also

_ModifiedMixin
ModifierManager

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

Calls 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. The decision logic is encapsulated within the self.fn property, which uses the _run_sync method to decide the execution mode.

Note

The self.fn property is responsible for selecting the appropriate execution path (sync or async) by leveraging the _run_sync method.

Parameters:

*args (~P) – Positional arguments to pass to the wrapped function.
**kwargs (~P) – Keyword arguments to pass to the wrapped function.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Return type:

Coroutine[Any, Any, T] | T

See also

default
_run_sync()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.ASyncFunctionAsyncDefault

Bases: ASyncFunction[P, T]

A specialized ASyncFunction that 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

@a_sync(default=’async’) async def my_function(x: int) -> str:

# Asynchronous call (default behavior) result = await my_function(5) # returns “5”

# Synchronous call result = my_function(5, sync=True) # returns “5”

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

Calls 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.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Returns:

The result of the function call.

Return type:

Coroutine[Any, Any, T] | T

See also

ASyncFunction.__call__()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.ASyncFunctionSyncDefault

Bases: ASyncFunction[P, T]

A specialized ASyncFunction that 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

@a_sync(default=’sync’) async def my_function(x: int) -> str:

# Synchronous call (default behavior) result = my_function(5) # returns “5”

# Asynchronous call result = await my_function(5, sync=False) # returns “5”

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

Calls 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.

Raises:

Exception – Any exception that may be raised by the wrapped function.

Returns:

The result of the function call.

Return type:

Coroutine[Any, Any, T] | T

See also

ASyncFunction.__call__()

__init__(_ModifiedMixin self, fn: AnyFn[P, T], _skip_validate: bint = False, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncFunction instance.

Parameters:

fn (AnyFn[P, T]) – The function to wrap.
_skip_validate (bint) – For internal use only. Skips validation of the wrapped function when its already been validated once before.
**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Return type:

None

See also

_validate_wrapped_fn()
ModifierManager

async all(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if all results of the function applied to the iterables are truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if all results are truthy, otherwise False.

Return type:

bint

See also

map()

async any(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → bint

Checks if any result of the function applied to the iterables is truthy.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

True if any result is truthy, otherwise False.

Return type:

bint

See also

map()

map(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → 'TaskMapping[P, T]'

Creates a TaskMapping for the wrapped function with the given iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

A TaskMapping object for managing concurrent execution.

Return type:

TaskMapping[P, T]

See also

TaskMapping

async max(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the maximum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

map()

async min(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Finds the minimum result of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

map()

async sum(self, *iterables: AnyIterable[Any], concurrency: Optional[int] = None, unicode task_name: str = u'', **function_kwargs: P.kwargs) → T

Calculates the sum of the results of the function applied to the iterables.

Parameters:

*iterables (AnyIterable[Any]) – Iterable objects to be used as arguments for the function.
concurrency (Optional[int]) – Optional maximum number of concurrent tasks.
task_name (str) – Optional name for the tasks.
**function_kwargs (P.kwargs) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

map()

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_fn = None

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.ASyncGenericBase

Bases: ASyncABC

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

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

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

Example

```python class MyClass(ASyncGenericBase):

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

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

Note

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

__init__(self)

class a_sync.a_sync.ASyncGenericSingleton[source]

Bases: ASyncGenericBase

A 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 ASyncSingletonMeta 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.

Note

This class can be instantiated directly, but it is intended to be subclassed to define specific asynchronous behavior. Subclasses should define the necessary properties and methods to specify the asynchronous behavior, as outlined in ASyncGenericBase.

Example

Create a subclass of ASyncGenericSingleton to define specific behavior:

class MyAsyncSingleton(ASyncGenericSingleton):
    @property
    def __a_sync_flag_name__(self):
        return "asynchronous"

    @property
    def __a_sync_flag_value__(self):
        return self.asynchronous

    @classmethod
    def __a_sync_default_mode__(cls):
        return False

    @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

See also

ASyncGenericBase for base functionality.
ASyncSingletonMeta for the metaclass managing the singleton behavior.

__init__(self)

class a_sync.a_sync.ASyncPropertyDescriptor

Bases: _ASyncPropertyDescriptorBase[I, T], AsyncPropertyDescriptor

Descriptor class for asynchronous properties.

__init__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

field_name

hidden_method_descriptor

hidden_method_name

max

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.ASyncPropertyDescriptorAsyncDefault

Bases: ASyncPropertyDescriptor[I, T]

A variant of ASyncPropertyDescriptor that 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__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all: ASyncFunctionAsyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionAsyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'async'

field_name

hidden_method_descriptor

hidden_method_name

max: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionAsyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.ASyncPropertyDescriptorSyncDefault

Bases: ASyncPropertyDescriptor[I, T]

A variant of ASyncPropertyDescriptor that 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__(_ModifiedMixin self, _fget: AsyncGetterFunction[I, T], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the _ASyncPropertyDescriptorBase.

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.

Return type:

None

async _all(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

awaitable_only(instance)

async get(self, instance: I, owner: Type[I] | None = None) → T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

get_loader(instance)

map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, int concurrency: Optional[int] = None, unicode name: str = u'') → 'TaskMapping[I, T]'

Maps the property across multiple instances.

Parameters:

instances (AnyIterable[I]) – An iterable of instances.
owner (Optional[Type[I]]) – The owner class of the property.
concurrency (Optional[int]) – Optional concurrency limit.
name (str) – Optional name for the task mapping.

Returns:

A TaskMapping object.

Return type:

_TaskMapping = None

all: ASyncFunctionSyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if all results are truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.all([1, 2, 3])

any: ASyncFunctionSyncDefault[AnyIterable[I], bool]

Create an ASyncFunction that checks if any result is truthy.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x > 0

instance = MyClass() result = await instance.my_method.any([-1, 0, 1])

default = 'sync'

field_name

hidden_method_descriptor

hidden_method_name

max: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the maximum result.

Returns:: An ASyncFunction object.

Examples

class MyClass:: @ASyncDescriptor def my_method(self, x):

return x

instance = MyClass() result = await instance.my_method.max([3, 1, 2])

min: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the minimum result.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.min([3, 1, 2]) ```

modifiers

sum: ASyncFunctionSyncDefault[AnyIterable[I], T]

Create an ASyncFunction that returns the sum of results.

Returns:: An ASyncFunction object.

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method.sum([1, 2, 3]) ```

class a_sync.a_sync.HiddenMethod

Bases: ASyncBoundMethodAsyncDefault[I, Tuple[()], T]

Represents a hidden method for asynchronous properties.

This class is used internally to manage hidden getter methods associated with a/sync properties.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

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

Call the bound method.

This method handles both synchronous and asynchronous calls based on the provided flags and the method’s configuration.

Parameters:

*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

Coroutine[Any, Any, T] | T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> await bound_method(arg1, arg2, kwarg1=value1)
>>> bound_method(arg1, arg2, kwarg1=value1, sync=True)

__init__(self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, unicode field_name: str, **modifiers: Unpack[ModifierKwargs]) → None

Initializes the HiddenMethod.

Parameters:

instance (I) – The instance to which the method is bound.
unbound (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function to be wrapped.
async_def (bool) – Indicates if the method is asynchronous.
field_name (str) – The name of the field associated with the method.
**modifiers (Unpack[ModifierKwargs]) – Additional modifier arguments.

Return type:

None

async all(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if all of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.all(iterable1, iterable2)

async any(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → bool

Check if any of the results are truthy.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.any(iterable1, iterable2)

map(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for this method.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Returns:

A TaskMapping instance for this method.

Return type:

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> task_mapping = bound_method.map(iterable1, iterable2, concurrency=5)
TODO briefly include how someone would then use task_mapping

async max(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.max(iterable1, iterable2)

async min(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.min(iterable1, iterable2)

async sum(self, *iterables: AnyIterable[I], concurrency: Optional[int] = None, unicode task_name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of the results.

Parameters:

*iterables (AnyIterable[I]) – Iterables to map over.
concurrency (Optional[int]) – Optional concurrency limit.
task_name (str) – Optional name for the task.
**kwargs (P.kwargs) – Additional keyword arguments.

Return type:

T

Examples

>>> bound_method = ASyncBoundMethod(instance, my_function, True)
>>> result = await bound_method.sum(iterable1, iterable2)

_async_def

Checks if the wrapped function is an asynchronous function.

Returns:: True if the function is asynchronous, otherwise False.

See also

asyncio.iscoroutinefunction()

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 wrapped function with async handling.

See also

_run_sync()
_await()

_asyncified

Converts the wrapped function to an asynchronous function and applies both sync and async modifiers.

Raises:: TypeError – If the wrapped function is already asynchronous.
Returns:: The asynchronous version of the wrapped function.

See also

_asyncify()

_cache_handle = None

_fn = None

_is_async_def

property _modified_fn: Callable[[P], Awaitable[T]] | Callable[[P], T]

Applies 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 modified function.

See also

ModifierManager.apply_async_modifiers()
ModifierManager.apply_sync_modifiers()

_sync_default

Determines 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 async.

See also

default

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 wrapped function with sync handling.

See also

_run_sync()
_asyncified()

default

DefaultMode

Gets the default execution mode (sync, async, or None) for the function.

Returns:: The default execution mode.

See also

ModifierManager.default

Type:: _ModifiedMixin.default

property fn

Returns the final wrapped version of ASyncFunction.__wrapped__ decorated with all of the a_sync goodness.

Returns:: The final wrapped function.

See also

_async_wrap()
_sync_wrap()

modifiers

class a_sync.a_sync.HiddenMethodDescriptor

Bases: ASyncMethodDescriptorAsyncDefault[I, Tuple[()], T]

Descriptor for hidden methods associated with asynchronous properties.

This class is used internally to manage hidden getter methods associated with a/sync properties.

Note

You should never create these yourself. They are automatically generated by ez-a-sync internally.

async __call__(self, instance: I, *args: P.args, **kwargs: P.kwargs) → T

Asynchronously call the method.

Parameters:

instance (I) – The instance the method is bound to.
*args (~P) – Positional arguments.
**kwargs (~P) – Keyword arguments.

Return type:

T

Examples

>>> descriptor = ASyncMethodDescriptor(my_async_function)
>>> await descriptor(instance, arg1, arg2, kwarg1=value1)

__init__(self, _fget: AnyFn[Concatenate[I, P], Awaitable[T]], unicode field_name: Optional[str] = None, **modifiers: Unpack[ModifierKwargs]) → None

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(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if all results are truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._all([1, 2, 3]) ```

async _any(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → bool

Check if any result is truthy.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._any([-1, 0, 1]) ```

async _max(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the maximum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._max([3, 1, 2]) ```

async _min(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Find the minimum result.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to check.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._min([3, 1, 2]) ```

async _sum(self, *instances: AnyIterable[I], concurrency: Optional[int] = None, unicode name: str = u'', **kwargs: P.kwargs) → T

Calculate the sum of results.

Parameters:

*instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to sum.
concurrency (int | None) – Optional maximum number of concurrent tasks.
name (str) – Optional name for the task.
**kwargs (~P) – Additional keyword arguments.

Return type:

T

Examples

```python class MyClass:

instance = MyClass() result = await instance.my_method._sum([1, 2, 3]) ```

map(self, *instances: AnyIterable[I], **bound_method_kwargs: P.kwargs) → 'TaskMapping[I, T]'

Create a TaskMapping for the given instances.

Parameters:

*instances (AnyIterable[I]) – Iterable of instances to map over.
**bound_method_kwargs (P.kwargs) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type: