a_sync.a_sync package

Subpackages

Submodules

a_sync.a_sync._descriptor module

This module contains the ASyncDescriptor class, which is used to create dual-function sync/async methods and properties.

The ASyncDescriptor class provides a base for creating descriptors that can handle both synchronous and asynchronous operations. It includes utility methods for mapping operations across multiple instances and provides access to common operations such as checking if all or any results are truthy, and finding the minimum, maximum, or sum of results of the method or property mapped across multiple instances through the use of ASyncFunction.

See also

class a_sync.a_sync._descriptor.ASyncDescriptor

Bases: _ModifiedMixin, Generic[I, P, T]

A descriptor base class for dual-function ASync methods and properties.

This class provides functionality for mapping operations across multiple instances and includes utility methods for common operations such as checking if all or any results are truthy, and finding the minimum, maximum, or sum of results of the method or property mapped across multiple instances through the use of ASyncFunction.

Examples

To create a dual-function method or property, subclass ASyncDescriptor and implement the desired functionality. You can then use the provided utility methods to perform operations across multiple instances.

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

See also

__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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

__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

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

Call self as a function.

Parameters:
__init__(name, bases, namespace)[source]
Parameters:
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):

@property def __a_sync_flag_name__(self) -> str:

return “sync”

@property def __a_sync_flag_value__(self) -> bool:

return True

@classmethod def __a_sync_default_mode__(cls) -> bool:

return False

```

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

def __init__(self, sync: bool):

self.sync = sync

@a_sync.property async def my_property(self):

return await some_async_operation()

@a_sync async def my_method(self):

return await another_async_operation()

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

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

Note

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

__init__(self)

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,

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

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.

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

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

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

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

Note

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

See also

ASyncFunction, ASyncDecorator

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

__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

__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

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(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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

_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

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

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

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:

return str(x)

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

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

__call__(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

__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

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(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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

_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

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

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

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:

return str(x)

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

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

__call__(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

__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

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(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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

_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

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

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

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

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

__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 (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.

  • async_def (bool) – Whether the original function is an async def.

  • **modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.

Return type:

None

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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:

TaskMapping[I, T]

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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

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

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

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

__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 (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.

  • async_def (bool) – Whether the original function is an async def.

  • **modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.

Return type:

None

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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:

TaskMapping[I, T]

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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

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

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

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

__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 (Callable[[Concatenate[I, ~P]], Awaitable[T]] | Callable[[Concatenate[I, ~P]], T]) – The unbound function.

  • async_def (bool) – Whether the original function is an async def.

  • **modifiers (Unpack[ModifierKwargs]) – Additional modifiers for the function.

Return type:

None

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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:

TaskMapping[I, T]

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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

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

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

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

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

class a_sync.a_sync.property.ASyncCachedPropertyDescriptorAsyncDefault

Bases: ASyncCachedPropertyDescriptor[I, T]

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

class a_sync.a_sync.property.ASyncCachedPropertyDescriptorSyncDefault

Bases: ASyncCachedPropertyDescriptor[I, T]

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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[I, T]

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

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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[I, T]

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

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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[I, T]

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

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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:

TaskMapping[I, T]

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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

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

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

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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[I, T]

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

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:
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:
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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

__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

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(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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The maximum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The minimum result.

Return type:

T

See also

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 (AsyncIterable[Any] | Iterable[Any]) – Iterable objects to be used as arguments for the function.

  • concurrency (int | None) – Optional maximum number of concurrent tasks.

  • task_name (str) – Optional name for the tasks.

  • **function_kwargs (~P) – Additional keyword arguments to pass to the function.

Returns:

The sum of the results.

Return type:

T

See also

_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

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

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

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

def __init__(self, sync: bool):

self.sync = sync

@a_sync.property async def my_property(self):

return await some_async_operation()

@a_sync async def my_method(self):

return await another_async_operation()

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

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

Note

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

__init__(self)
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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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[I, T]

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

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – Additional keyword arguments.

Return type:

bool

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:

TaskMapping[I, T]

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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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 (AsyncIterable[I] | Iterable[I]) – Iterables to map over.

  • concurrency (int | None) – Optional concurrency limit.

  • task_name (str) – Optional name for the task.

  • **kwargs (~P) – 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

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

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

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

bool

Examples

```python class MyClass:

@ASyncDescriptor def my_method(self, x):

return x > 0

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

TaskMapping[I, T]

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:

@ASyncDescriptor def my_method(self, x):

return x

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:

@ASyncDescriptor def my_method(self, x):

return x

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

a_sync.a_sync.cached_property

alias of ASyncCachedPropertyDescriptor

a_sync.a_sync.property

alias of ASyncPropertyDescriptor

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

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

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

Parameters:

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

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

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

Return type:

ASyncDecorator | ASyncFunction[~P, T]

Modifiers:

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

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

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

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

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

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

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

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

Examples

The decorator can be used in several ways.

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

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

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

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

Note

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

See also

ASyncFunction, ASyncDecorator