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[source]

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__(_fget, field_name=None, **modifiers)[source]

Initialize the ASyncDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)[source]

Check if all results are truthy.

Parameters:

  • *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(*instances, concurrency=None, name='', **kwargs)[source]

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)[source]

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(*instances, concurrency=None, name='', **kwargs)[source]

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(*instances, concurrency=None, name='', **kwargs)[source]

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(*instances, **bound_method_kwargs)[source]

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped

a_sync.a_sync._flags module

This module provides functionality for handling synchronous and asynchronous flags in the ez-a-sync library.

ez-a-sync uses ‘flags’ to indicate whether objects 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._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_legacy(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.flags.AFFIRMATIVE_FLAGS: Set[str] = {'sync'}

Set of flags indicating synchronous behavior.

This set currently contains only the flag “sync”, which is used to denote synchronous operations within the ez-a-sync library.

Examples

>>> 'sync' in AFFIRMATIVE_FLAGS
True

>>> 'async' in AFFIRMATIVE_FLAGS
False

See also

NEGATIVE_FLAGS: Flags indicating asynchronous behavior. VIABLE_FLAGS: All valid flags, combining both sync and async indicators.

a_sync.a_sync.flags.NEGATIVE_FLAGS: Set[str] = {'asynchronous'}

Set of flags indicating asynchronous behavior.

This set currently contains only the flag “asynchronous”, which is used to denote asynchronous operations within the ez-a-sync library.

Examples

>>> 'asynchronous' in NEGATIVE_FLAGS
True

>>> 'sync' in NEGATIVE_FLAGS
False

See also

AFFIRMATIVE_FLAGS: Flags indicating synchronous behavior. VIABLE_FLAGS: All valid flags, combining both sync and async indicators.

a_sync.a_sync.flags.VIABLE_FLAGS: Set[str] = {'asynchronous', 'sync'}

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

The ez-a-sync library uses these 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.

Examples

>>> 'sync' in VIABLE_FLAGS
True

>>> 'asynchronous' in VIABLE_FLAGS
True

>>> 'invalid' in VIABLE_FLAGS
False

See also

AFFIRMATIVE_FLAGS: Flags indicating synchronous behavior. NEGATIVE_FLAGS: Flags indicating asynchronous behavior.

a_sync.a_sync.function module

class a_sync.a_sync.function.ASyncDecorator

Bases: _ModifiedMixin

__call__(self, func: AnyFn[P, T]) → ASyncFunction[P, T]

Decorates a function with async or sync behavior based on the default modifier.

Parameters:

func (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to decorate.

Returns:

An ASyncFunction instance with the appropriate default behavior.

Return type:

ASyncFunction[~P, T]

See also

__init__(self, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncDecorator instance.

Parameters:

**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

validate_inputs(self) → None

Validates the input modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager.default

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

modifiers: ModifierManager
wrapped
class a_sync.a_sync.function.ASyncDecoratorAsyncDefault

Bases: ASyncDecorator

__call__(self, func: AnyFn[P, T]) → ASyncFunctionAsyncDefault[P, T]
Parameters:

func (Callable[[~P], Awaitable[T]] | Callable[[~P], T])

Return type:

ASyncFunctionAsyncDefault[~P, T]

__init__(self, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncDecorator instance.

Parameters:

**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

validate_inputs(self) → None

Validates the input modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager.default

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

modifiers: ModifierManager
wrapped
class a_sync.a_sync.function.ASyncDecoratorSyncDefault

Bases: ASyncDecorator

__call__(self, func: AnyFn[P, T]) → ASyncFunctionSyncDefault[P, T]
Parameters:

func (Callable[[~P], Awaitable[T]] | Callable[[~P], T])

Return type:

ASyncFunctionSyncDefault[~P, T]

__init__(self, **modifiers: Unpack[ModifierKwargs]) → None

Initializes an ASyncDecorator instance.

Parameters:

**modifiers (Unpack[ModifierKwargs]) – Keyword arguments for function modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

validate_inputs(self) → None

Validates the input modifiers.

Raises:

ValueError – If ‘default’ is not ‘sync’, ‘async’, or None.

Return type:

None

See also

  • ModifierManager.default

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

modifiers: ModifierManager
wrapped
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__(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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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

property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

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

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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

property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

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

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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

property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

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

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

modifiers: ModifierManager
wrapped
a_sync.a_sync.function._check_not_genfunc(func: Callable) → None

Raises an error if the function is a generator or async generator.

Parameters:

func (Callable) – The function to check.

Raises:

ValueError – If the function is a generator or async generator.

Return type:

None

See also

a_sync.a_sync.function._validate_argspec(fn: Callable)
Parameters:

fn (Callable)

a_sync.a_sync.function._validate_wrapped_fn(fn: Callable) → None

Ensures ‘fn’ is an appropriate function for wrapping with a_sync.

Parameters:

fn (Callable) – The function to validate.

Raises:

  • TypeError – If the input is not callable.

  • RuntimeError – If the function has arguments with names that conflict with viable flags.

Return type:

None

See also

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__(self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:
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)
_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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)
property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

_cache_handle: TimerHandle = 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()

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:
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)
_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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)
property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

_cache_handle: TimerHandle = 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()

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(self, instance: I, unbound: AnyFn[Concatenate[I, P], T], async_def: bool, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the bound method.

Parameters:
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)
_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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)
property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

_cache_handle: TimerHandle = 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()

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(_fget, field_name=None, **modifiers)

Initialize the ASyncDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, **bound_method_kwargs)

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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__(_fget, field_name=None, **modifiers)

Initialize the ASyncDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, **bound_method_kwargs)

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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__(_fget, field_name=None, **modifiers)

Initialize the ASyncDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, **bound_method_kwargs)

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.ASyncCachedPropertyDescriptorAsyncDefault

Bases: cached_property[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.

__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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.ASyncCachedPropertyDescriptorSyncDefault

Bases: cached_property[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.

__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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.ASyncPropertyDescriptor

Bases: _ASyncPropertyDescriptorBase[I, T], AsyncPropertyDescriptor

Descriptor class for asynchronous properties.

__init__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.ASyncPropertyDescriptorAsyncDefault

Bases: property[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__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.ASyncPropertyDescriptorSyncDefault

Bases: property[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__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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 methods associated with asynchronous properties.

__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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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)
property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

_cache_handle: TimerHandle = 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()

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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 methods associated with asynchronous properties.

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]], field_name: str | None = None, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the HiddenMethodDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, **bound_method_kwargs)

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor: HiddenMethodDescriptor[T]
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.cached_property

Bases: ASyncCachedPropertyDescriptor[I, T]

Descriptor for defining cached properties that can be accessed both synchronously and asynchronously.

__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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property.property

Bases: ASyncPropertyDescriptor[I, T]

Descriptor for defining properties that can be accessed both synchronously and asynchronously.

__init__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
a_sync.a_sync.property._parse_args(func: None | DefaultMode | AsyncGetterFunction[I, T], modifiers: ModifierKwargs) → Tuple[AsyncGetterFunction[I, T] | None, ModifierKwargs]

Parses the arguments for the property decorators.

Parameters:

  • func (None | Literal['sync', 'async', None] | ~typing.Callable[[~a_sync._typing.I], ~typing.Awaitable[~a_sync._typing.T]]) – The function to be wrapped.

  • modifiers (ModifierKwargs) – Additional modifier arguments.

Returns:

A tuple containing the parsed function and modifiers.

Return type:

Tuple[Callable[[I], Awaitable[T]] | None, ModifierKwargs]

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]], cached_property[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncCachedPropertyDescriptorSyncDefault[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncCachedPropertyDescriptorAsyncDefault[I, T]]

a_sync.a_sync.property.a_sync_property(func: 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]], property[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncPropertyDescriptorSyncDefault[I, T]] | Callable[[Callable[[I], Awaitable[T]] | Callable[[I], T]], ASyncPropertyDescriptorAsyncDefault[I, T]]

a_sync.a_sync.singleton module

class a_sync.a_sync.singleton.ASyncGenericSingleton[source]

Bases: 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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__(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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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

property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

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

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
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 methods associated with asynchronous properties.

__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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

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)
property _async_def: bint

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

property _asyncified: Callable[[P], Awaitable[T]]

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

_cache_handle: TimerHandle = 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()

property _sync_default: bint

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

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

property fn

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

Returns:

The final wrapped function.

See also

modifiers: ModifierManager
wrapped
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 methods associated with asynchronous properties.

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]], field_name: str | None = None, **modifiers: Unpack[ModifierKwargs]) → None

Initialize the HiddenMethodDescriptor.

Parameters:
Raises:

ValueError – If _fget is not callable.

Return type:

None

async _all(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, **bound_method_kwargs)

Create a TaskMapping for the given instances.

Parameters:

  • *instances (AsyncIterable[I] | Iterable[I]) – Iterable of instances to map over.

  • **bound_method_kwargs (~P) – Additional keyword arguments for the bound method.

Returns:

A TaskMapping object.

Return type:

TaskMapping[I, T]

Examples

class MyClass:

@ASyncDescriptor def my_method(self, x):

return x * 2

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

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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

The name of the field the ASyncDescriptor is bound to.

property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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: ModifierManager
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.cached_property

Bases: ASyncCachedPropertyDescriptor[I, T]

Descriptor for defining cached properties that can be accessed both synchronously and asynchronously.

__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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

_check_method_name(method, method_type)
_check_method_sync(method, method_type)
async _max(*instances, concurrency=None, name='', **kwargs)

Find the maximum result.

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

already_loaded(instance)
del_cache_value(instance)
deleter(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(instance)
get_cache_value(instance)
get_instance_state(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) → 'asyncio.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(instance)
map(self, instances: AnyIterable[I], owner: Optional[Type[I]] = None, 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]

not_loaded(instance)
pop_lock(self, instance: I) → None

Removes the lock for the property.

Parameters:

instance (I) – The instance from which the property is accessed.

Return type:

None

set_cache_value(instance, value)
setter(method)
property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

wrapped
class a_sync.a_sync.property

Bases: ASyncPropertyDescriptor[I, T]

Descriptor for defining properties that can be accessed both synchronously and asynchronously.

__init__(self, _fget: AsyncGetterFunction[I, T], field_name: str | None = 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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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

_asyncify(self, func: SyncFn[P, T]) → CoroFn[P, T]

Converts a synchronous function to an asynchronous one and applies async modifiers.

Parameters:

func (Callable[[~P], T]) – The synchronous function to be converted.

Returns:

The asynchronous version of the function with applied modifiers.

Return type:

Callable[[~P], Awaitable[T]]

See also

  • ModifierManager.apply_async_modifiers()

async _max(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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(*instances, concurrency=None, name='', **kwargs)

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, 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]

property _TaskMapping: Type[TaskMapping]

This silly helper just fixes a circular import

property _await: Callable[[Awaitable[T]], T]

Applies sync modifiers to the _helpers._await function and caches it.

Returns:

The modified _await function.

See also

  • ModifierManager.apply_sync_modifiers()

property all: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property any: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property default: Literal['sync', 'async', None]

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

Returns:

The default execution mode.

See also

  • ModifierManager.default

field_name

The name of the field the ASyncDescriptor is bound to.

hidden_method_descriptor
hidden_method_name
property max: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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])

property min: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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
property sum: ASyncFunction[Concatenate[AsyncIterable[I] | Iterable[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]) ```

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

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

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

Parameters:

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

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

  • **modifiers (Unpack[ModifierKwargs]) – Additional keyword arguments to modify the behavior of the decorated function. See 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