BaseNotifier#

class BaseNotifier(channel=None, mention_to=None, mention_level='error', mention_if_ends=True, callsite_level='error', token=None, verbose=True, disable=False)[source]#

Bases: ABC

Abstract base class for all notifiers.

Provides common functionality for sending messages and watching code execution, with optional exception handling and verbosity.

__init__(channel=None, mention_to=None, mention_level='error', mention_if_ends=True, callsite_level='error', token=None, verbose=True, disable=False)[source]#

Initialize the notifier with default settings. This settings can be overridden at each call of register(), send(), and watch().

Parameters:

  • channel (str | None) – Default channel for notifications. If not provided, it will look for an environment variable named {platform}_CHANNEL where {platform} is the notifier’s platform name in uppercase (e.g., SLACK_CHANNEL for Slack).

  • mention_to (str | None) – Default user to mention in notification. If not provided, it will look for an environment variable named {platform}_MENTION_TO where {platform} is the notifier’s platform name in uppercase (e.g., SLACK_MENTION_TO for Slack).

  • mention_level (Literal['info', 'warning', 'error']) – Minimum log level to trigger a mention.

  • mention_if_ends (bool) – Whether to mention at the end of the watch.

  • callsite_level (Literal['info', 'warning', 'error']) – Minimum log level to emit the call-site source snippet.

  • token (str | None) – API token or authentication key. If not provided, it will look for an environment variable named {platform}_BOT_TOKEN where {platform} is the notifier’s platform name in uppercase (e.g., SLACK_BOT_TOKEN for Slack).

  • verbose (bool | int) – If obj:True, log messages to console. If set to 1, only logs during initialization. If set to 2 or higher, behaves the same as obj:True.

  • disable (bool) – If True, disable sending all notifications. This is useful for parallel runs or testing where you want to avoid sending actual notifications.

Note

The channel and token must be set, either via environment variables or as function arguments. If not set, the notification will not be sent, and an error will be logged (the original Python script will continue running without interruption).

register(target, name, params=None, **options)[source]#

Register existing function or method to be monitored by this notifier. This function corresponds to applying the watch() decorator to an existing function or method.

Parameters:

  • target (ModuleType | type[Any] | Any) – The module, class, or class instance containing the function to be registered.

  • name (str) – The name of the function to be registered.

  • params (str | list[str] | None) – Names of the function parameters whose values should be included in the message when the registered function is called.

  • **options (Unpack[SendOptions]) – Additional options. See SendOptions for details.

Return type:

None

send(data, *, channel=None, mention_to=None, verbose=None, disable=None)[source]#

Send a notification message. You can send notifications at any point in your code, not just at the start or end of a task. Any data can be sent, and it will be stringified.

Parameters:

  • data (Any) – The payload or message content.

  • channel (str | None) – Override the default channel for notifications.

  • mention_to (str | None) – Override the default entity to mention on notification.

  • verbose (bool | None) – Override the default verbosity setting.

  • disable (bool | None) – Override the default disable flag.

Return type:

None

watch(iterable=None, /, *, params=None, step=1, total=None, **options)[source]#

If iterable is not provided, return an object that can serve as both a context manager and a decorator to watch code execution. This will automatically send notifications when the function or code block starts, ends, or raises an exception.

If iterable is provided, return a generator that yields items from an iterable while sending notifications about its progress.

Parameters:

  • iterable (Iterable[TypeVar(T)] | None) – An iterable (e.g., a list or range) to monitor progress.

  • params (str | list[str] | None) – Names of the function parameters whose values should be included in the message when the decorated function is called. This option is ignored when used as a context manager.

  • step (int) – The number of items to process before sending a progress notification. This option is ignored if the iterable is not provided.

  • total (int | None) – The total number of items in the iterable. If not provided and the iterable has not __len__, it will not be included in the progress messages. This option is ignored if the iterable is not provided.

  • **options (Unpack[SendOptions]) – Additional options. See SendOptions for details.

Return type:

ContextManagerDecorator | ContextManagerIterator[TypeVar(T)]

Returns:

An an object that can serve as both a context manager and a decorator.

class ContextManagerDecorator(*args, **kwargs)[source]#

Bases: Protocol[F]

Protocol for objects that can be used as context managers and decorators.

__init__(*args, **kwargs)#
class ContextManagerIterator(*args, **kwargs)[source]#

Bases: Protocol[T_co]

Protocol for objects that can be used as context managers and iterators.

__init__(*args, **kwargs)#
typeddict SendOptions[source]#

Bases: TypedDict

Additional options for watch() and register().

Optional Keys:

  • label (Optional[str]) – Optional label for the message. Included in notification messages and log entries.

  • channel (Optional[str]) – Override the default channel for notifications.

  • mention_to (Optional[str]) – Override the default user to mention on notification.

  • mention_level (Optional[Literal['info', 'warning', 'error']]) – Override the default mention threshold level.

  • mention_if_ends (Optional[bool]) – Override the default setting for whether to mention at the end of the watch.

  • callsite_level (Optional[Literal['info', 'warning', 'error']]) – Override the default call-site source snippet threshold level.

  • callsite_context_before (int) – Number of lines of context to include before the call site. Default is 1.

  • callsite_context_after (int) – Number of lines of context to include after the call site. Default is 4.

  • verbose (Optional[bool]) – Override the default verbosity setting.

  • disable (Optional[bool]) – Override the default disable flag.