server.asyncio_extensions
-async def map_suppress(func: Callable[[~T], Coroutine[Any, Any, Any]], iterable: Iterable[~T], logger: logging.Logger = <Logger server.asyncio_extensions (WARNING)>, msg: str = '')
+async def map_suppress(func: Callable[[~T], Coroutine[Any, Any, Any]],
iterable: Iterable[~T],
logger: logging.Logger = <Logger server.asyncio_extensions (WARNING)>,
msg: str = '')
async def map_suppress(
+ func: Callable[[T], Coroutine[Any, Any, Any]],
+ iterable: Iterable[T],
+ logger: logging.Logger = logger,
+ msg: str = ""
+):
+ results = await asyncio.gather(
+ *(func(item) for item in iterable),
+ return_exceptions=True
+ )
+ for result, item in zip(results, iterable):
+ if isinstance(result, BaseException):
+ logger.exception(
+ "Unexpected error %s%s",
+ msg,
+ item,
+ exc_info=result
+ )
def synchronized(*args)
def synchronized(*args):
+ """
+ Ensure that a function will only execute in serial.
+
+ # Params
+ - `lock`: An instance of asyncio.Lock to use for synchronization.
+ """
+ # Invoked like @synchronized
+ if args and inspect.isfunction(args[0]):
+ return _synchronize(args[0])
+
+ # Invoked like @synchronized() or @synchronized(args, ...)
+ return _partial(_synchronize, *args)Ensure that a function will only execute in serial.
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.
+ """
+ # Invoked like @synchronizedmethod
+ if args and inspect.isfunction(args[0]):
+ return _synchronize_method(args[0])
+
+ # Invoked like @synchronizedmethod() or @synchronizedmethod(args, ...)
+ return _partial(_synchronize_method, *args)Create a method that will be wrapped with an async lock.
class AsyncLock(Protocol, AsyncContextManager["AsyncLock"]):
+ def locked(self) -> bool: ...
+ async def acquire(self) -> bool: ...
+ def release(self) -> None: ...Base class for protocol classes.
Protocol classes are defined as::
class Proto(Protocol):
@@ -97,15 +178,6 @@ Classes
def meth(self) -> T:
...
class AsyncLock(Protocol, AsyncContextManager["AsyncLock"]):
- def locked(self) -> bool: ...
- async def acquire(self) -> bool: ...
- def release(self) -> None: ...async def acquire(self) -> bool: ...
def locked(self) ‑> bool
def locked(self) -> bool: ...
def release(self) ‑> None
def release(self) -> None: ...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.
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.
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 TrueSleeps 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."""
+ return self._lockedReturn True if lock is acquired.
def release(self) ‑> None
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.")When invoked on an unlocked lock, a RuntimeError is raised.
-Generated by pdoc 0.11.1.
+Generated by pdoc 0.11.5.