a_sync.primitives.locks package
Submodules
a_sync.primitives.locks.counter module
This module provides two specialized async flow management classes, CounterLock
and CounterLockCluster
.
These primitives manage synchronization of tasks that must wait for an internal counter to reach a specific value.
- class a_sync.primitives.locks.counter.CounterLock[source]
Bases:
_DebugDaemonMixin
An async primitive that uses an internal counter to manage task synchronization.
A coroutine can await counter.wait_for(3) and it will wait until the internal counter >= 3. If some other task executes counter.value = 5 or counter.set(5), the first coroutine will proceed as 5 >= 3.
The internal counter can only be set to a value greater than the current value.
See also
CounterLockCluster
for managing multipleCounterLock
instances.- __init__(start_value=0, name=None)[source]
Initializes the
CounterLock
with a starting value and an optional name.- Parameters:
Examples
>>> counter = CounterLock(start_value=0, name="example_counter") >>> counter.value 0
- async _debug_daemon()[source]
Periodically logs debug information about the counter state and waiters.
This method is used internally to provide debugging information when debug logging is enabled.
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- set(value)[source]
Sets the counter to the specified value.
This method internally uses the value property to enforce that the new value must be strictly greater than the current value.
- Parameters:
value (int) – The value to set the counter to. Must be strictly greater than the current value.
- Raises:
ValueError – If the new value is less than or equal to the current value.
- Return type:
None
Examples
>>> counter = CounterLock(start_value=0) >>> counter.set(5) >>> counter.value 5
See also
CounterLock.value()
for direct value assignment.
- async wait_for(value)[source]
Waits until the counter reaches or exceeds the specified value.
This method will ensure the debug daemon is running if the counter is not ready.
Examples
>>> counter = CounterLock(start_value=0) >>> await counter.wait_for(5) # This will block until counter.value >= 5
See also
CounterLock.set()
to set the counter value.
- _daemon
- _events: DefaultDict[int, Event]
A defaultdict that maps each awaited value to an
Event
that manages the waiters for that value.
- _name
An optional name for the counter, used in debug logs.
- _value
The current value of the counter.
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- is_ready
A lambda function that indicates whether the current counter value is greater than or equal to a given value.
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- class a_sync.primitives.locks.counter.CounterLockCluster[source]
Bases:
object
An asyncio primitive that represents a collection of
CounterLock
objects.wait_for(i) will wait until the value of all
CounterLock
objects is >= i.See also
CounterLock
for managing individual counters.- __init__(counter_locks)[source]
Initializes the
CounterLockCluster
with a collection ofCounterLock
objects.- Parameters:
counter_locks (Iterable[CounterLock]) – The
CounterLock
objects to manage.- Return type:
None
Examples
>>> lock1 = CounterLock(start_value=0) >>> lock2 = CounterLock(start_value=0) >>> cluster = CounterLockCluster([lock1, lock2])
- async wait_for(value)[source]
Waits until the value of all
CounterLock
objects in the cluster reaches or exceeds the specified value.Examples
>>> lock1 = CounterLock(start_value=0) >>> lock2 = CounterLock(start_value=0) >>> cluster = CounterLockCluster([lock1, lock2]) >>> await cluster.wait_for(5) # This will block until all locks have value >= 5
- locks
a_sync.primitives.locks.event module
This module provides an enhanced version of asyncio.Event with additional debug logging to help detect deadlocks.
- class a_sync.primitives.locks.event.Event[source]
Bases:
Event
,_DebugDaemonMixin
An asyncio.Event with additional debug logging to help detect deadlocks.
This event class extends asyncio.Event by adding debug logging capabilities. It logs detailed information about the event state and waiters, which can be useful for diagnosing and debugging potential deadlocks.
- __init__(name='', debug_daemon_interval=300, *, loop=None)[source]
Initializes the Event.
- Parameters:
name (str) – An optional name for the event, used in debug logs.
debug_daemon_interval (int) – The interval in seconds for the debug daemon to log information.
loop (Optional[asyncio.AbstractEventLoop]) – The event loop to use.
- async _debug_daemon()[source]
Periodically logs debug information about the event state and waiters.
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- clear()[source]
Reset the internal flag to false. Subsequently, coroutines calling wait() will block until set() is called to set the internal flag to true again.
- set()[source]
Set the internal flag to true. All coroutines waiting for it to become true are awakened. Coroutine that call wait() once the flag is true will not block at all.
- async wait()[source]
Wait until the event is set.
- Returns:
True when the event is set.
- Return type:
Literal[True]
- _daemon
- _debug_daemon_interval
- _loop: AbstractEventLoop = None
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
a_sync.primitives.locks.prio_semaphore module
This module provides priority-based semaphore implementations. These semaphores allow waiters to be assigned priorities, ensuring that higher priority waiters are processed before lower priority ones.
- class a_sync.primitives.locks.prio_semaphore.Priority[source]
Bases:
Protocol
- __init__(*args, **kwargs)
- class a_sync.primitives.locks.prio_semaphore.PrioritySemaphore[source]
Bases:
_AbstractPrioritySemaphore
[int
|float
|Decimal
,_PrioritySemaphoreContextManager
]Semaphore that uses numeric priorities for waiters.
This class extends
_AbstractPrioritySemaphore
and provides a concrete implementation using numeric priorities. The _context_manager_class is set to_PrioritySemaphoreContextManager
, and the _top_priority is set to -1, which is the highest priority.Examples
The primary way to use this semaphore is by specifying a priority.
>>> priority_semaphore = PrioritySemaphore(10) >>> async with priority_semaphore[priority]: ... await do_stuff()
You can also enter and exit this semaphore without specifying a priority, and it will use the top priority by default:
>>> priority_semaphore = PrioritySemaphore(10) >>> async with priority_semaphore: ... await do_stuff()
See also
_AbstractPrioritySemaphore
for the base class implementation.- _context_manager_class
alias of
_PrioritySemaphoreContextManager
- __call__(fn)
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __getitem__(priority)
Gets the context manager for a given priority.
- Parameters:
priority (PT | None) – The priority for which to get the context manager. If None, uses the top priority.
- Returns:
The context manager associated with the given priority.
- Return type:
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> context_manager = semaphore[priority]
- __init__(value=1, *, name=None)
Initializes the priority semaphore.
- Parameters:
- Return type:
None
Examples
>>> semaphore = _AbstractPrioritySemaphore(5, name="test_semaphore")
- _count_waiters()
Counts the number of waiters for each priority.
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore._count_waiters()
- async _debug_daemon()
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _wake_up_next()
Wakes up the next waiter in line.
This method handles the waking of waiters based on priority. It includes an emergency procedure to handle potential lost waiters, ensuring that no waiter is left indefinitely waiting.
The emergency procedure is a temporary measure to address potential issues with lost waiters.
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore._wake_up_next()
- Return type:
None
- async acquire()
Acquires the semaphore with the top priority.
This method overrides
Semaphore.acquire()
to handle priority-based logic.Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> await semaphore.acquire()
- Return type:
Literal[True]
- decorate(fn)
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- locked()
Checks if the semaphore is locked.
- Returns:
True if the semaphore cannot be acquired immediately, False otherwise.
- Return type:
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore.locked()
- release()
Release a semaphore, incrementing the internal counter by one.
When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.
- _capacity
The initial capacity of the semaphore.
- _context_managers: Dict[PT, _AbstractPrioritySemaphoreContextManager[PT]]
A dictionary mapping priorities to their context managers.
- _daemon
- _loop = None
- _potential_lost_waiters: List[Future[None]]
A list of futures representing waiters that might have been lost.
- _top_priority = -1
- _waiters: List[_AbstractPrioritySemaphoreContextManager[PT]]
A heap queue of context managers, sorted by priority.
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- class a_sync.primitives.locks.prio_semaphore._AbstractPrioritySemaphore[source]
Bases:
Semaphore
,Generic
[PT
,CM
]A semaphore that allows prioritization of waiters.
This semaphore manages waiters with associated priorities, ensuring that waiters with higher priorities are processed before those with lower priorities. Subclasses must define the _top_priority attribute to specify the default top priority behavior.
The _context_manager_class attribute should specify the class used for managing semaphore contexts.
See also
PrioritySemaphore
for an implementation using numeric priorities.- __call__(fn)
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __getitem__(priority)[source]
Gets the context manager for a given priority.
- Parameters:
priority (PT | None) – The priority for which to get the context manager. If None, uses the top priority.
- Returns:
The context manager associated with the given priority.
- Return type:
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> context_manager = semaphore[priority]
- __init__(value=1, *, name=None)[source]
Initializes the priority semaphore.
- Parameters:
- Return type:
None
Examples
>>> semaphore = _AbstractPrioritySemaphore(5, name="test_semaphore")
- _count_waiters()[source]
Counts the number of waiters for each priority.
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore._count_waiters()
- async _debug_daemon()
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _wake_up_next()[source]
Wakes up the next waiter in line.
This method handles the waking of waiters based on priority. It includes an emergency procedure to handle potential lost waiters, ensuring that no waiter is left indefinitely waiting.
The emergency procedure is a temporary measure to address potential issues with lost waiters.
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore._wake_up_next()
- Return type:
None
- async acquire()[source]
Acquires the semaphore with the top priority.
This method overrides
Semaphore.acquire()
to handle priority-based logic.Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> await semaphore.acquire()
- Return type:
Literal[True]
- decorate(fn)
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- locked()[source]
Checks if the semaphore is locked.
- Returns:
True if the semaphore cannot be acquired immediately, False otherwise.
- Return type:
Examples
>>> semaphore = _AbstractPrioritySemaphore(5) >>> semaphore.locked()
- release()
Release a semaphore, incrementing the internal counter by one.
When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.
- _capacity
The initial capacity of the semaphore.
- property _context_manager_class: Type[_AbstractPrioritySemaphoreContextManager[PT]]
- _context_managers: Dict[PT, _AbstractPrioritySemaphoreContextManager[PT]]
A dictionary mapping priorities to their context managers.
- _daemon
- _loop = None
- _potential_lost_waiters: List[Future[None]]
A list of futures representing waiters that might have been lost.
- property _top_priority: PT
Defines the top priority for the semaphore.
Subclasses must implement this property to specify the default top priority.
- Raises:
NotImplementedError – If not implemented in a subclass.
- _waiters: List[_AbstractPrioritySemaphoreContextManager[PT]]
A heap queue of context managers, sorted by priority.
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- class a_sync.primitives.locks.prio_semaphore._AbstractPrioritySemaphoreContextManager[source]
-
A context manager for priority semaphore waiters.
This context manager is associated with a specific priority and handles the acquisition and release of the semaphore for waiters with that priority.
- __call__(fn)
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __init__(parent, priority, name=None)[source]
Initializes the context manager for a specific priority.
- Parameters:
parent (_AbstractPrioritySemaphore) – The parent semaphore.
priority (PT) – The priority associated with this context manager.
name (str | None) – An optional name for the context manager, used for debugging.
- Return type:
None
Examples
>>> parent_semaphore = _AbstractPrioritySemaphore(5) >>> context_manager = _AbstractPrioritySemaphoreContextManager(parent_semaphore, priority=1)
- async _debug_daemon()
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _repr_no_parent_()[source]
Returns a string representation of the context manager without the parent.
- Return type:
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _wake_up_next()
Wake up the first waiter that isn’t done.
- async acquire()[source]
Acquires the semaphore for this context manager.
If the internal counter is larger than zero on entry, decrement it by one and return True immediately. If it is zero on entry, block, waiting until some other coroutine has called release() to make it larger than 0, and then return True.
This method overrides
Semaphore.acquire()
to handle priority-based logic.Examples
>>> context_manager = _AbstractPrioritySemaphoreContextManager(parent, priority=1) >>> await context_manager.acquire()
- Return type:
Literal[True]
- decorate(fn)
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- locked()
Returns True if semaphore cannot be acquired immediately.
- release()[source]
Releases the semaphore for this context manager.
This method overrides
Semaphore.release()
to handle priority-based logic.Examples
>>> context_manager = _AbstractPrioritySemaphoreContextManager(parent, priority=1) >>> context_manager.release()
- Return type:
None
- _daemon
- _loop: AbstractEventLoop = None
- _parent
The parent semaphore.
- _priority
The priority associated with this context manager.
- _value
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- property loop: AbstractEventLoop
Gets the event loop associated with this context manager.
- name
- class a_sync.primitives.locks.prio_semaphore._PrioritySemaphoreContextManager[source]
Bases:
_AbstractPrioritySemaphoreContextManager
[int
|float
|Decimal
]Context manager for numeric priority semaphores.
- __call__(fn)
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __init__(parent, priority, name=None)
Initializes the context manager for a specific priority.
- Parameters:
parent (_AbstractPrioritySemaphore) – The parent semaphore.
priority (PT) – The priority associated with this context manager.
name (str | None) – An optional name for the context manager, used for debugging.
- Return type:
None
Examples
>>> parent_semaphore = _AbstractPrioritySemaphore(5) >>> context_manager = _AbstractPrioritySemaphoreContextManager(parent_semaphore, priority=1)
- async _debug_daemon()
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _repr_no_parent_()
Returns a string representation of the context manager without the parent.
- Return type:
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _wake_up_next()
Wake up the first waiter that isn’t done.
- async acquire()
Acquires the semaphore for this context manager.
If the internal counter is larger than zero on entry, decrement it by one and return True immediately. If it is zero on entry, block, waiting until some other coroutine has called release() to make it larger than 0, and then return True.
This method overrides
Semaphore.acquire()
to handle priority-based logic.Examples
>>> context_manager = _AbstractPrioritySemaphoreContextManager(parent, priority=1) >>> await context_manager.acquire()
- Return type:
Literal[True]
- decorate(fn)
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- locked()
Returns True if semaphore cannot be acquired immediately.
- release()
Releases the semaphore for this context manager.
This method overrides
Semaphore.release()
to handle priority-based logic.Examples
>>> context_manager = _AbstractPrioritySemaphoreContextManager(parent, priority=1) >>> context_manager.release()
- Return type:
None
- _daemon
- _loop: AbstractEventLoop = None
- _parent
The parent semaphore.
- _priority
The priority associated with this context manager.
- _priority_name = 'priority'
- _value
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- property loop: AbstractEventLoop
Gets the event loop associated with this context manager.
- name
a_sync.primitives.locks.semaphore module
This module provides various semaphore implementations, including a debug-enabled semaphore, a dummy semaphore that does nothing, and a threadsafe semaphore for use in multi-threaded applications.
- class a_sync.primitives.locks.semaphore.DummySemaphore[source]
Bases:
Semaphore
A dummy semaphore that implements the standard
asyncio.Semaphore
API but does nothing.This class is useful for scenarios where a semaphore interface is required but no actual synchronization is needed.
Example
dummy_semaphore = DummySemaphore()
- async def no_op():
- async with dummy_semaphore:
return 1
- __init__(name=None)[source]
Initialize the dummy semaphore with an optional name.
- Parameters:
name (optional) – An optional name for the dummy semaphore.
- _get_loop()
- _wake_up_next()
Wake up the first waiter that isn’t done.
- locked()
Returns True if semaphore cannot be acquired immediately.
- _loop = None
- _value
- name
- class a_sync.primitives.locks.semaphore.Semaphore[source]
Bases:
Semaphore
,_DebugDaemonMixin
A semaphore with additional debugging capabilities inherited from
_DebugDaemonMixin
.This semaphore includes debug logging capabilities that are activated when the semaphore has waiters. It allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
You can write this pattern:
``` semaphore = Semaphore(5)
- async def limited():
- async with semaphore:
return 1
like this:
``` semaphore = Semaphore(5)
@semaphore async def limited():
return 1
See also
_DebugDaemonMixin
for more details on debugging capabilities.- __call__(fn)[source]
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __init__(value, name=None, **kwargs)[source]
Initialize the semaphore with a given value and optional name for debugging.
- Parameters:
value (int) – The initial value for the semaphore.
name (optional) – An optional name used only to provide useful context in debug logs.
- Return type:
None
- async _debug_daemon()[source]
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- async acquire()[source]
Acquire the semaphore, ensuring that debug logging is enabled if there are waiters.
If the semaphore value is zero or less, the debug daemon is started to log the state of the semaphore.
- Returns:
True when the semaphore is successfully acquired.
- Return type:
Literal[True]
- decorate(fn)[source]
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- release()[source]
Release a semaphore, incrementing the internal counter by one.
When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.
- _daemon
- _loop = None
- _value
- _waiters
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- name
- class a_sync.primitives.locks.semaphore.ThreadsafeSemaphore[source]
Bases:
Semaphore
A semaphore that works in a multi-threaded environment.
This semaphore ensures that the program functions correctly even when used with multiple event loops. It provides a workaround for edge cases involving multiple threads and event loops by using a separate semaphore for each thread.
Example
semaphore = ThreadsafeSemaphore(5)
- async def limited():
- async with semaphore:
return 1
See also
Semaphore
for the base class implementation.- __call__(fn)
Decorator method to wrap coroutine functions with the semaphore.
This allows rewriting the pattern of acquiring a semaphore within a coroutine using a decorator.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- __init__(value, name=None)[source]
Initialize the threadsafe semaphore with a given value and optional name.
- Parameters:
value (int | None) – The initial value for the semaphore, should be an integer.
name (optional) – An optional name for the semaphore.
- Return type:
None
- async _debug_daemon()
Daemon coroutine (runs in a background task) which will emit a debug log every minute while the semaphore has waiters.
This method is part of the
_DebugDaemonMixin
and is used to provide detailed logging information about the semaphore’s state when it is being waited on.Example
semaphore = Semaphore(5)
- async def monitor():
await semaphore._debug_daemon()
- Return type:
None
- _ensure_debug_daemon(*args, **kwargs)
Ensures that the debug daemon task is running.
This method checks if the debug daemon is already running and starts it if necessary. If debug logging is not enabled, it sets the daemon to a dummy future.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
Either the debug daemon task or a dummy future if debug logging is not enabled.
- Return type:
Future[None]
Examples
Ensuring the debug daemon is running:
my_instance = MyDebugClass() my_instance._ensure_debug_daemon()
See also
_start_debug_daemon()
for starting the daemon.
- _get_loop()
- _start_debug_daemon(*args, **kwargs)
Starts the debug daemon task if debug logging is enabled and the event loop is running.
This method checks if debug logging is enabled and if the event loop is running. If both conditions are met, it starts the debug daemon task.
- Parameters:
*args – Positional arguments for the debug daemon.
**kwargs – Keyword arguments for the debug daemon.
- Returns:
The debug daemon task as an asyncio.Task, or a dummy future if debug logs are not enabled or if the daemon cannot be created.
- Return type:
Future[None]
Examples
Starting the debug daemon:
my_instance = MyDebugClass() my_instance._start_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _stop_debug_daemon(t=None)
Stops the debug daemon task.
This method cancels the debug daemon task if it is running. Raises a ValueError if the task to be stopped is not the current daemon.
- Parameters:
t (optional) – The task to be stopped, if any.
- Raises:
ValueError – If t is not the current daemon.
- Return type:
None
Examples
Stopping the debug daemon:
my_instance = MyDebugClass() my_instance._stop_debug_daemon()
See also
_ensure_debug_daemon()
for ensuring the daemon is running.
- _wake_up_next()
Wake up the first waiter that isn’t done.
- async acquire()
Acquire the semaphore, ensuring that debug logging is enabled if there are waiters.
If the semaphore value is zero or less, the debug daemon is started to log the state of the semaphore.
- Returns:
True when the semaphore is successfully acquired.
- Return type:
Literal[True]
- decorate(fn)
Wrap a coroutine function to ensure it runs with the semaphore.
Example
semaphore = Semaphore(5)
@semaphore async def limited():
return 1
- locked()
Returns True if semaphore cannot be acquired immediately.
- release()
Release a semaphore, incrementing the internal counter by one.
When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.
- _daemon
- _loop = None
- _value
- _waiters
- property debug_logs_enabled: bool
Checks if debug logging is enabled for the logger.
Examples
>>> class MyClass(_LoggerMixin): ... pass ... >>> instance = MyClass() >>> instance.debug_logs_enabled False
See also
- dummy
- property logger: Logger
Provides a logger instance specific to the class using this mixin.
The logger ID is constructed from the module and class name, and optionally includes an instance name if available.
Examples
>>> class MyClass(_LoggerMixin): ... _name = "example" ... >>> instance = MyClass() >>> logger = instance.logger >>> logger.name 'module_name.MyClass.example'
>>> class AnotherClass(_LoggerMixin): ... pass ... >>> another_instance = AnotherClass() >>> another_logger = another_instance.logger >>> another_logger.name 'module_name.AnotherClass'
Note
Replace module_name with the actual module name where the class is defined.
See also
- name
- property semaphore: Semaphore
Returns the appropriate semaphore for the current thread.
NOTE: We can’t cache this property because we need to check the current thread every time we access it.
Example
semaphore = ThreadsafeSemaphore(5)
- async def limited():
- async with semaphore.semaphore:
return 1
- semaphores: DefaultDict[Thread, Semaphore]