Pythonプロセス終了時に最後のエラーを取得しチャットワークに通知する

6877 ワード
エラー Python 運用 保守 Chatwork Python テキストリンク

やりたいこと

定期バッチ処理が正常に終了したかどうかをchatworkに通知できれば保守運用が楽になると考えました。
要素としては以下の組み合わせですが、

  • プロセス終了時に決まった処理を行いたい
  • 最後に発生したエラーの種類を取得したい
  • try-exceptで地道に処理するのではなく、一箇所の記述でグローバルに行いたい

これらをPythonで実現するにはどうしたらよいかをまとめました。
最後にchatworkに通知するサンプルコードを記載しました。

プロセス終了時に決まった処理を行いたい

プロセス終了時にフックして処理を行いたいときはatexitを利用できます。
今回のように「終了時に通知を送りたい」とか「終了時に一時ファイルを確実に消したい」といったケースで利用できるでしょう。
atexit.register(func)で関数を登録できます。

最後に発生したエラーの種類を取得したい

Pythonではsys.last_typeで最後に発生したエラーを取得できます。
Rubyでいう$!、PHPでいうerror_get_lastです。

通常は定義されておらず、捕捉されない例外が発生してインタプリタがエラーメッセージとトレースバックを出力した場合にのみ設定されます。

とドキュメントにあるように例外が起こらなかった場合はsys.last_typeがAttributeErrorになります。
例外がtry-exceptなどで補足されていた場合も同様です。
よってhasattr(sys, "last_type")などでAttributeErrorが起こらないか前方確認が必要です。

これらを組み合わせて

import sys
import atexit

def some_function_at_exit():
    if hasattr(sys, "last_type"):
        foo()
    else:
        bar()

atexit.register(some_function_at_exit)

異常終了したときはfoo()、正常終了したときはbar()が実行される記述です。
if hasattr(sys, "last_type"):としたのですべてのエラーにフックしますが、特定のエラーのみにフックさせたいときはif sys.last_type == ZeroDivisionError:のように書き換えればよいです。

chatworkに通知を送るサンプルコード

import os
import sys
import atexit

import chatpywork

def chatwork_notify(text):
    server_name = os.uname()[1]
    project = os.path.basename(os.getcwd())
    process_name = sys.argv[0]

    title = f"[title]{project}: {process_name} on {server_name}[/title]"
    body = f"[info]{title}{text}[/info]"

    room_id = os.environ["CHATWORK_ROOM_ID"]
    api_key = os.environ["CHATWORK_API_KEY"]
    room = chatpywork.Room(room_id, api_key)
    room.send_message(body)

def chatwork_notify_at_exit():
    if hasattr(sys, "last_type"):
        chatwork_notify(f"(devil)(devil)(devil){sys.last_type.__name__}: プロセスは異常終了しました。(devil)(devil)(devil)")
    else:
        chatwork_notify("プロセスは正常終了しました。")

atexit.register(chatwork_notify_at_exit)

↓実行結果