SlackNotifier#

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

Bases: BaseNotifier

__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) – 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).

Example

from notist import SlackNotifier

# Create a SlackNotifier with defaults
slack = SlackNotifier(
    channel="my-channel",  # Slack channel name or ID
    mention_to="@U012345678",  # Mention a specific user (Optional)
)
register(target, name, params=None, **options)#

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

Example

Monitor existing functions from libraries:

import requests

# Register the `get` function from the `requests` library
slack.register(requests, "get")

# Now any time you call `requests.get`, it will be monitored
response = requests.get("https://example.com/largefile.zip")

Monitor existing methods of classes:

from transformers import Trainer

# Register the `train` method of the `Trainer` class
slack.register(Trainer, "train")

# Now any time you call `trainer.train()`, it will be monitored
trainer = Trainer(model=...)
trainer.train()

Monitor existing methods of specific class instances:

from transformers import Trainer

# Create a Trainer instance
trainer = Trainer(model=...)

# Register the `train` method of the `trainer` instance
slack.register(trainer, "train")

# Now any time you call `trainer.train()`, it will be monitored
trainer.train()
send(data, *, channel=None, mention_to=None, verbose=None, disable=None)#

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

Example

# Immediately send "Job finished!" to your Slack channel
slack.send("Job finished!")

# You can also send any Python data (it will be stringified)
slack.send(data)
watch(iterable=None, /, *, params=None, step=1, total=None, **options)#

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.

Example

Monitor functions:

# You can also optionally specify params to include in the notification
# The values passed to these parameters are also reported
@slack.watch(params=["arg1", "arg2"])
def long_task(arg1: int, arg2: str, arg3: bool) -> None:
    # This function will be monitored
    # You can receive notifications when it starts, ends, or encounters an error
    ...
    # Your long-running code here

Monitor methods:

with slack.watch():
    # Code inside this block will be monitored
    # You can receive notifications when it starts, ends, or encounters an error
    ...
    # Your long-running code here

Monitor Iterations (e.g., for loops):

for i in slack.watch(range(100), step=10):
    # This loop will be monitored, and you'll receive notifications every 10 iterations.
    ...
    # Your long-running code here

Note

The above example does not catch exceptions automatically, since exceptions raised inside the for loop cannot be caught by the iterator in Python. If you also want to be notified when an error occurs, wrap your code in the monitoring context:

with slack.watch(range(100), step=10) as it:
    for i in it:
        # This loop will be monitored, and you'll receive notifications every 10 iterations.
        # If an error occurs inside this context, you'll be notified immediately.
        ...
        # Your long-running code here