Qiita Teams that are logged in
You are not logged in to any team

Log in to Qiita Team
Community
OrganizationEventAdvent CalendarQiitadon (β)
Service
Qiita JobsQiita ZineQiita Blog
4
Help us understand the problem. What are the problem?

posted at

updated at

Organization

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)

↓実行結果
スクリーンショット 2021-02-19 13.03.55.png

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
4
Help us understand the problem. What are the problem?