Python コードの実行状態を Slack や Discord に簡単に通知できるライブラリを作った

はじめに

機械学習モデルの訓練など、時間がかかるプログラムを実行した時に、それが問題なく完了したのか・エラーになって止まってしまったのかなど、プログラムの実行状態を知りたい場合がしばしばあります。しかし、定期的にターミナルやログファイルを開いて自分から確認するのはやや手間なので、スマホや PC で簡単に通知を受け取りたい気がします。

そこで、実行状態の通知を簡単に受け取れるライブラリを作成しました。自分用に作ったものですが、ドキュメント等も整備して公開しているので、もし使いたい方がいればどうぞです：

このライブラリを使うと、Python プログラム中の関数・コードブロック・for 文などが開始・終了・エラーに遭遇した際に簡単に通知を受け取ることが出来ます。現状は Slack と Discord に対応しています。新たな機能要望やバグ遭遇などあれば上記の GitHub へ issue/pull request をお送りください。

もしこのライブラリを便利に感じたら、上記 GitHub にスターをつけてくださると嬉しいです

使い方

インストール方法

$ pip install notist

Slack bot や Discord bot のセットアップ

Slack や Discord で通知を受け取れるようにするために、Slack bot や Discord bot を作成し、そのトークンを環境変数などに設定する必要があります。詳しい設定方法はクイックスタートなどを参照してください。

主な機能の使い方

詳細な使い方は API リファレンスなどを参照してください

デフォルトの設定と初期化

まず初めに、送信先やチャンネルなどの情報を設定し初期化します：

import notist

# Set up Slack notifiers with defaults
notist.init(send_to="slack", channel="my-channel")

この操作は必須ではないですが、これを行わなかった場合、後続の呼び出しの際に同様の設定をする必要があります。

関数・コードブロック・for 文などをモニタリングする

notist.watch 関数を使用すると、関数・コードブロック・for 文などを簡単にモニタリングすることが出来ます。

関数のモニタリング

関数定義の際にデコレータとして付けることで、この関数が開始・終了・エラーに遭遇した際に通知を受け取れます。また、オプショナルな params 引数に、デコレートしている関数の引数の名前を入れることで、それがどのような値とともに呼び出されたのかも通知から確認可能になります：

@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

コードブロックのモニタリング

with 文で囲むことで、任意の位置でコードブロックをモニタリング可能です：

with notist.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

for 文などのイテレーションのモニタリング

イテラブルなオブジェクトを囲むことで、その進行度合いを通知で確認出来ます。また、オプショナルな step 引数を指定することで、何ステップごとに通知を受け取るか指定可能です：

for i in notist.watch(range(100), step=10):
   # This loop will be monitored, and you'll receive notifications every 10 iterations
   ...

上記のようなコードを実行すると、以下のような通知を受け取ることが可能です：

• 関数が開始した際：

  Start watching <function `__main__.without_error`>
   ▷ Defined at: /home/kaito47802/workspace/NotifyState/sample.py:21
   ▷ Called from: `__main__` @ /home/kaito47802/workspace/NotifyState/sample.py:28
  

• 関数が終了した際：

  End watching <function `__main__.without_error`>
   ▷ Defined at: /home/kaito47802/workspace/NotifyState/sample.py:21
   ▷ Called from: `__main__` @ /home/kaito47802/workspace/NotifyState/sample.py:28
   ⦿ Execution time: 0s
  

• 関数がエラーに遭遇した際：

  @kAIto47802
  Error while watching <function `__main__.with_error`>
   ▷ Defined at: /home/kaito47802/workspace/NotifyState/sample.py:15
   ▷ Called from: `__main__` @ /home/kaito47802/workspace/NotifyState/sample.py:30
    29 │     print("Example function that raises an error")
    30 │     with_error()
  ╭───────┄┄ ────────────
  │ 31 │     print("You will see a Slack notification for the error above")
  │ 32 │     print(
  │ 33 │         "You can use the watch() helper as a function decorator or as a context manager"
  ╰─❯ Exception: This is an error
   ⦿ Execution time: 0s
  
  > Traceback (most recent call last):
  >  File "/home/kaito47802/.pyenv/versions/3.12.0/lib/python3.12/contextlib.py", line 81,   in inner
  >    return func(*args, **kwds)
  >           ^^^^^^^^^^^^^^^^^^^
  >  File "/home/kaito47802/workspace/NotifyState/sample.py", line 18, in with_error
  >    raise Exception("This is an error")
  > Exception: This is an error
  

上記の for 文などのイテレーションのモニタリングの例ではエラーの遭遇は通知されません。これは for 文のループの中で起きたエラーはイテレータ自身からは捕捉出来ないためです。各ステップの進行度合いだけでなく、エラーの場合も通知したければ、以下のように with 文と組み合わせる必要があります：

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

既存の関数やメソッドをモニタリングする

notist.register 関数を使用することで、上記のような自分のコードだけでなく、外部ライブラリの実行状態もモニタリング可能です。

外部ライブラリの既存の関数のモニタリング

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

既存クラスのメソッドのモニタリング

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

既存クラスの特定のインスタンスのメソッドのモニタリング

from transformers import Trainer

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

# Register the `train` method of the `trainer` instance
# This will not affect other instances of Trainer
notist.register(trainer, "train")

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

単発の通知を送信する

上記のような場合だけでなく、notist.send 関数を使用すれば任意の場所で好きな内容を通知できます：

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

通知クラスを利用する

また、上記のような呼び出し方だけでなく、通知クラスのインスタンスを作成して使用することも可能です：

from notist import SlackNotifier

# Create a SlackNotifier with defaults
slack = SlackNotifier(
    channel="my-channel",
    mention_to="@U012345678",  # Mention a specific user (Optional)
)

# Watch a function:
@slack.watch()
def long_task():
    ...
    # Your long-running code here

終わりに

自分は機械学習関連の研究を主にしており、その実験の際にこのような通知を簡単に受け取りたいと感じたため、作成するに至りました。自分用に作ったものですが、これが誰かの役にも立つなら幸いです。

また、初期リリース (v0.1.0) では未実装ですが、以下のような機能なども今後のバージョンで実装しようかなとうっすらと考えています：

• CLI での呼び出しを可能にし、任意のプログラムの開始・終了・終了コードを通知する