はじめに
大規模なデータ分析やLLMのチューニングなどを行う際にプログラムの実行時間は何十時間にも及ぶことがある。byobuやtmuxなどで実行していると実行の完了に気づけなかったり、予期しないエラーで実行が停止している場合があった。
そこで、プログラムの実行やエラーをdiscordやteams等の普段使っているツールに通知するデコレータを作成し、自作のライブラリ化した。
あくまでも個人で使用するツールの開発備忘録です。もっと良い実装や汎用性の高い実装なども考えられます。
作ったもの
以下のように利用できるデコレータを作成した。 target_functionの実行が完了するかエラーが発生した場合に引数で指定したツールに通知を飛ばす実装である。
@monitoring(platform="teams", webhook_url=WEBHOOK_URL)
def target_function():
...
実行完了時に以下のような通知を受け取ることができる。
デコレータ本体は以下のように実装している。platformごとに送信処理を分けることで必要に応じて様々なプラットフォームを追加することができる。(現在はdiscordとteamsに対応)
import traceback
import time
import requests
from functools import wraps
from typing import Callable, Literal
def monitoring(platform: Literal["discord", "teams"], webhook_url: str) -> Callable:
"""
関数の実行時間の計測と実行・失敗状況を監視するデコレータ。
処理終了時およびエラー時に指定されたプラットフォームへ通知を送信する。
Args:
platform: 通知先のプラットフォーム. "discord" または "teams" を指定。
webhook_url: 通知先のWebhook URL.
Example:
```python
@monitoring(platform="discord", webhook_url="~~~discord_webhook_url~~~")
def hoge_function():
...
```
```python
@monitoring(platform="teams", webhook_url="~~~teams_webhook_url~~~")
def hoge_function():
...
```
"""
def _send_discord_message(message: str):
requests.post(webhook_url, json={"content": message})
def _send_teams_message(message: str):
payload = {
"type": "message",
"attachments": [
{
"contentType": "application/vnd.microsoft.card.adaptive",
"contentUrl": None,
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.4",
"body": [
{
"type": "TextBlock",
"text": message,
"wrap": True,
}
],
},
}
],
}
requests.post(webhook_url, json=payload)
_send_message_exporter = {
"discord": _send_discord_message,
"teams": _send_teams_message,
}
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
elapsed_time = time.time() - start_time
hours = int(elapsed_time // 3600)
minutes = int((elapsed_time % 3600) // 60)
seconds = int(elapsed_time % 60)
time_str = f"{hours}時間{minutes}分{seconds}秒"
message = f"✅ 実行が正常に完了しました\n関数: {func.__name__}\n実行時間: {time_str} ({elapsed_time:.2f}秒)"
_send_message_exporter[platform](message)
return result
except Exception as e:
elapsed_time = time.time() - start_time
traceback_str = ''.join(traceback.format_exception(e))
print(traceback_str)
hours = int(elapsed_time // 3600)
minutes = int((elapsed_time % 3600) // 60)
seconds = int(elapsed_time % 60)
time_str = f"{hours}時間{minutes}分{seconds}秒"
message = f"❌ エラーが発生しました: {e}\n関数: {func.__name__}\n実行時間: {time_str} ({elapsed_time:.2f}秒)\nスタックトレース:\n{traceback_str}"
_send_message_exporter[platform](message)
raise e
return wrapper
return decorator
また、github上にライブラリとして公開することによって以下のコマンドで追加することができるようにした。
pip install git+https://you22fy/my_python_utils
今後の展望として、slack, Line等の複数のプラットフォームを追加することを検討している。