Core Utils#

init(*, send_to, 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(). Alternatively, you can skip initialization with this function and provide all settings directly through these functions.

Parameters:

  • send_to (Literal['slack', 'discord'] | list[Literal['slack', 'discord']]) – Destination(s) to send notifications to. e.g., “slack”, “discord”,

  • ["slack" (or)

  • "discord"].

  • 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.

Return type:

None

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

Note

The destination (send_to) must be set, either in this init() function or as an argument to subsequent calls.

Example

import notist

# Set up Slack notifiers with defaults
notist.init(send_to="slack", channel="my-channel", mention_to="@U012345678")
register(target, name, params=None, *, send_to=None, **options)[source]#

Register existing function or method to be watched 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.

  • send_to (Literal['slack', 'discord'] | list[Literal['slack', 'discord']] | None) – Destination(s) to send notifications to. e.g., “slack”, “discord”, or [“slack”, “discord”].

  • **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
notist.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
notist.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
notist.register(trainer, "train")

# Now any time you call `trainer.train()`, it will be monitored
trainer.train()
send(data, *, send_to=None, 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.

  • send_to (Literal['slack', 'discord'] | list[Literal['slack', 'discord']] | None) – Destination(s) to send notifications to. e.g., “slack”, “discord”, or [“slack”, “discord”].

  • 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
notist.send("Job finished!")

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

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.

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 parameters or variables whose values should be included in the message. When used as a decorator, these names are resolved against the decorated function’s arguments. When used as a context manager, they are resolved against the current scope at the call site.

  • label – Optional label for the watch context. This label will be included in both notification messages and log entries.

  • send_to (Literal['slack', 'discord'] | list[Literal['slack', 'discord']] | None) – Destination(s) to send notifications to. e.g., “slack”, “discord”, or [“slack”, “discord”].

  • **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

Use as a decorator to monitor a function:

# You can also optionally specify params to include in the notification
# The values passed to these parameters are also reported
@notist.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

# You can also use it to monitor async functions
@notist.watch()
async def long_task_async() -> None:
    # This async function will be monitored
    # You can receive notifications when it starts, ends, or encounters an error
    ...
    # Your long-running code here

Use as a context manager to monitor a block of code:

var1, var2 = 123, "abc"
# You can also optionally specify params to include in the notification.
# The values of variables with these names in the current scope are also reported.
with notist.watch(params=["var1", "var2"]):
    # Code inside this block will be monitored
    # You can receive notifications when it starts, ends, or encounters an error
    ...
    # Your long-running code here

Use to monitor an iterable in a for loop:

for i in notist.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 for monitoring iterations 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 notist.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