asyncutils.signals

Functions related to asynchronous signal handling.

Functions

wait_for_signal(…)

Module Contents

async asyncutils.signals.wait_for_signal[T](
processor: collections.abc.Callable[[signal.Signals], collections.abc.Awaitable[T]],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[True],
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
default_on_processor_failure: T,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T[source]
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], collections.abc.Awaitable[T]],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[True],
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], collections.abc.Awaitable[T]],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[False] = ...,
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], collections.abc.Awaitable[T]],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[False] = ...,
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
default_on_processor_failure: T,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], T],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[True],
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
default_on_processor_failure: T,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], T],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[True],
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], T],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[False] = ...,
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
async asyncutils.signals.wait_for_signal(
processor: collections.abc.Callable[[signal.Signals], T],
/,
*S: int,
timeout: float | None = ...,
raise_on_timeout: Literal[False] = ...,
loop: asyncio.events.AbstractEventLoop | None = ...,
possible_errors: tuple[asyncutils._internal.types.ExcType, Ellipsis] = ...,
default_on_processor_failure: T,
sigs: collections.abc.Iterable[int] = ...,
logger: logging.Logger = ...,
) → T | None
Wait for an operating system level signal included in sigs (default context.WAIT_FOR_SIGNAL_DEFAULT_SIGNALS) and the variable
positional arguments to be signaled within timeout and handle it.
See the docs for the signal module, add_signal_handler(), as well as the Wikipedia page for signals.
processor should be a function that takes the signal occurred, preferrably returning an awaitable object.
If raise_on_timeout is True, throw TimeoutError on timeout. Otherwise, return None.
If loop is passed, its add_signal_handler() and remove_signal_handler() methods will be used; a loop is created and set otherwise.
Errors whose types are included in possible_errors will cause the logger logger to emit an error and the function to return
default_on_processor_failure, or None if not passed.
Some information related to the progress of the wait also goes to the logger.
The return value of the processor is returned through this function.
Note

There is limited support for signals on Windows and this function may not work as expected even with the little signals it provides. Therefore, the function emits a warning unless the console is running, in which case it would be much too annoying.