pythonの実行完了をslackに通知する時の私的ベストプラクティス
pythonで重い処理をしている場合に、終わったらslackに通知を送れるように自分用のライブラリを作ったので紹介します。
この方法では、実行ファイルに 2行追加するだけで 通知が実現できます。
概要
以下の手順で実装しました。
- slackアプリを作成する
- slackアプリからの通知を実装する
- 通知送信をデコレータとして実装する
- ライブラリとして使えるように整備する
前準備:slackアプリの作成
以下の手順で作成し、アクセストークンを発行しておきます。
- slack api(https://api.slack.com/) からアプリを作成
- Permissions -> Scopeから必要な権限を追加
- slackワークスペースにアプリをインストール
- Bot User OAuth Access TokenのAPI tokenを取得
必要な権限は以下の通り。
- Bot Token Scopes
- chat:write
- chat:write.public
実装
使用ライブラリ
pip install slack_sdk
pip install python-dotenv # 環境変数の読み込みに使用
フォルダ構成
libs
└─ slackapp
├─ .env
├─ __init__.py
└─ main.py
libsフォルダにパスを通しておきます。そうすれば他のディレクトリで作業していてもslackappをライブラリとして呼び出せます。
他にも自作ライブラリがあればここに入れましょう。
ソースコード
SLACK_API_TOKEN=xoxb-111-222-xxxxx
アクセストークンは、実行ファイル内にベタ打ちせずに環境変数に設定することが推奨されます。
いろいろやり方はありますが、私はpython-dotenv パッケージ を利用するやり方が手軽で気に入ってます。
下にある main.py の中で load_dotenv を呼び出すと、.envにある変数を環境変数と同様に呼べるようになります。
from slackapp.main import *
__init__.pyについて
__init__.py は中身空っぽのままでもいいんですが、↑のようにしておくと、他でインポートして使用する時にモジュール名を省略できます。
import slackapp
# __init__.py が空っぽの場合
@slackapp.main.notify
def ufunc():
return
# __init__.py を書いた場合
@slackapp.notify
def ufunc():
return
詳細は こちらのQiita記事 に説明があります。
import os
from datetime import datetime
import functools
from dotenv import load_dotenv
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
load_dotenv()
def notify(channel="notice"):
def _notify(func):
@functools.wraps(func)
def _wrapper(*args, **kwargs):
start_time = datetime.now()
text = f'\nfunction: {func.__name__}\nstart at {start_time:%Y/%m/%d %H:%M:%S}\nelapsed '
try:
result = func(*args, **kwargs)
except BaseException as e:
end_time = datetime.now()
text = f':warning: {e.__class__.__name__}: {e}' + text + format_timedelta(end_time - start_time)
send_slack_message(text=text, channel=channel)
raise e
else:
end_time = datetime.now()
text = ':white_check_mark: Finished!' + text + format_timedelta(end_time - start_time)
send_slack_message(text=text, channel=channel)
return result
return _wrapper
return _notify
def send_slack_message(text="Hello from slackapp! :tada:", channel="random"):
slack_token = os.environ["SLACK_API_TOKEN"]
client = WebClient(token=slack_token)
try:
response = client.chat_postMessage(
channel=channel,
text=text,
)
except SlackApiError as e:
# You will get a SlackApiError if "ok" is False
assert e.response["error"]
def format_timedelta(timedelta):
total_sec = timedelta.total_seconds()
fmt = ''
# hours
if total_sec > 3600:
hours = total_sec // 3600
fmt += f'{hours:.0f}h '
# minutes
if total_sec > 60:
minutes = (total_sec % 3600) // 60
fmt += f'{minutes:.0f}m '
# seconds
seconds = (total_sec % 60) // 1
fmt += f'{seconds:.0f}s'
elif total_sec > 1:
fmt = f'{total_sec:.3f} s'
else:
milliseconds = total_sec * 1000
fmt = f'{milliseconds:.0f} ms'
return fmt
使い方
以下のサンプルコードのように、自作ライブラリをインポートして通知したい関数にデコレータをつけると、処理が終わった時にFinishedの通知が来ます。
import slackapp
@slackapp.notify
def decorator_test():
for _ in range(10**5):
pass
return
decorator_test()
途中でエラーが出た場合は、その内容が通知メッセージにも含まれます。
import slackapp
@slackapp.notify
def decorator_test_zerodivision():
for _ in range(10**5):
pass
return 1 / 0 # ZeroDivisionError
decorator_test_zerodivision()
Ephemeral メッセージを使う
Ephemeral メッセージとは、「あなただけに表示されています」と表示されているメッセージのことです。「Ephemeral」は「つかの間の」という意味で、その名の通り一定時間が経過すればそのメッセージは表示されなくなるそうです。[1]
実行終了を通知したいだけであれば、エファメラルメッセージにして送る方法でも問題ない場合が多いと思います。
Ephemeral メッセージを使う場合は、以下2点に留意して、main.py の send_slack_message を修正します。
- ユーザーID（Uxxxxxxx 形式のもの）を指定すること。
slackアプリのプロフィールから「メンバーIDをコピー」で確認できます。 - 送信先のチャンネルに、自分が入っていること。
def send_slack_message(text="Hello from slackapp! :tada:", channel="random"):
slack_token = os.environ["SLACK_API_TOKEN"]
client = WebClient(token=slack_token)
user_id = 'Uxxxxxxx' # 自分のユーザーIDに置き換える
try:
response = client.chat_postEphemeral( # chat_postMessage の代わり
channel=channel,
text=text,
user=user_id, # ユーザーIDを指定する
)
except SlackApiError as e:
# You will get a SlackApiError if "ok" is False
assert e.response["error"]
おわりに
もっといい方法があるよって場合はコメントで教えてください。
参考
-
参考記事にそのような記述がありましたが、いつ消えるのかよくわからなかったです。 ↩︎
Discussion