Some helper functions for common async tasks

Functions

async def map_suppress(func: Callable[[~T], Coroutine[Any, Any, Any]], iterable: Iterable[~T], logger: logging.Logger = <Logger server.asyncio_extensions (WARNING)>, msg: str = '')

def synchronized(*args)

Ensure that a function will only execute in serial.

Params

• lock: An instance of asyncio.Lock to use for synchronization.

def synchronizedmethod(*args)

Create a method that will be wrapped with an async lock.

Params

• attrname: The name of the lock attribute that will be used. If the attribute doesn't exist or is None, a lock will be created. The default is to use a value based on the decorated function name.

Classes

class AsyncLock (*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing), for example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

Expand source code

class AsyncLock(Protocol, AsyncContextManager["AsyncLock"]):
    def locked(self) -> bool: ...
    async def acquire(self) -> bool: ...
    def release(self) -> None: ...

Ancestors

• typing.Protocol
• contextlib.AbstractAsyncContextManager
• typing.Generic
• abc.ABC

Methods

async def acquire(self) ‑> bool

def locked(self) ‑> bool

def release(self) ‑> None

class SpinLock (sleep_duration: float = 0.01)

An asyncio spinlock. The advantage of using this over asyncio.Lock is that it can be called accross multiple event loops at the cost of being less performant. As with any spinlock, it's best used in situations where concurrent access is unlikely.

Expand source code

class SpinLock(_ContextManagerMixin):
    """
    An asyncio spinlock. The advantage of using this over asyncio.Lock is that
    it can be called accross multiple event loops at the cost of being less
    performant. As with any spinlock, it's best used in situations where
    concurrent access is unlikely.
    """

    def __init__(self, sleep_duration: float = 0.01):
        self.sleep_duration = sleep_duration
        self._locked = False

    def __repr__(self) -> str:
        res = super().__repr__()
        extra = "locked" if self._locked else "unlocked"
        return f"<{res[1:-1]} [{extra}]>"

    def locked(self) -> bool:
        """Return True if lock is acquired."""
        return self._locked

    async def acquire(self) -> bool:
        """
        Sleeps repeatedly for sleep_duration until the lock is unlocked, then
        sets it to locked and returns True.
        """
        while self._locked:
            await asyncio.sleep(self.sleep_duration)

        self._locked = True
        return True

    def release(self) -> None:
        """
        When invoked on an unlocked lock, a RuntimeError is raised.
        """
        if self._locked:
            self._locked = False
        else:
            raise RuntimeError("Lock is not acquired.")

Ancestors

• asyncio.locks._ContextManagerMixin

Methods

async def acquire(self) ‑> bool

Sleeps repeatedly for sleep_duration until the lock is unlocked, then sets it to locked and returns True.

def locked(self) ‑> bool

Return True if lock is acquired.

def release(self) ‑> None

When invoked on an unlocked lock, a RuntimeError is raised.