Jupyter Notebook で例外が発生した際に通知を行う方法

みなさんは、 Jupyter Notebook で機械学習モデルの訓練等の長時間かかるコードを実行している際に、
いつの間にかエラーが発生して止まってしまっていた、ということはありませんか？

今回は、そのような例外(エラー)が発生した際に Slack や Discord に自動で通知を行う方法を紹介します。

通知用の関数を作成する

初めに、通知を行うための関数を用意しましょう。
Slack や Discord であれば以下のような関数で通知を行えます。

• Slack用通知スクリプト

  import requests
  
  
  def slack_notify(message, webhook_url):
      """Slack通知を行う関数."""
      requests.post(webhook_url, json={"text": message})
  

• Discord用通知スクリプト

  import requests
  
  
  def discord_notify(message, webhook_url):
      """Discord通知を行う関数."""
      requests.post(webhook_url, json={"content": message})
  

特定のセルで例外(エラー)が発生した際にスクリプトを実行する

特定のセルで例外が発生した場合に通知を行う方法は簡単で、以下のようにtry構文を用いて実現できます。

try:
    <実行する内容>
except Exception:
    slack_notify("ERROR: 例外が発生しました。", "<WEBHOOK URL>")

任意のセルで例外(エラー)が発生した際にスクリプトを実行する

特定のセルではなく、全てのセルで例外が発生した場合に通知したい場合、以下の方法で実現できます。

set_custom_exc という関数を使用すると、例外時の処理を追加できます。
以下のコードは stackoverflowの回答 を参考に作成しています(ほぼ同じです)。

from IPython import get_ipython
from IPython.core.ultratb import AutoFormattedTB


def jupyter_preprocess():
    """Jupyter Notebook の最初に実行する関数."""
    # ref. https://stackoverflow.com/a/40135960
    itb = AutoFormattedTB(mode='Plain', tb_offset=1)
    def custom_exc(shell, etype, evalue, tb, tb_offset=None):
        shell.showtraceback((etype, evalue, tb), tb_offset=tb_offset)
        stb = itb.structured_traceback(etype, evalue, tb)
        sstb = itb.stb2text(stb)

        # ここに例外発生時に実行するコードを書く
        slack_notify("ERROR: 例外が発生しました。", "<WEBHOOK URL>")

        return sstb

    get_ipython().set_custom_exc((Exception,), custom_exc)

このように定義した jupyter_preprocess を最初に実行しておくことで、例外(エラー)発生時に通知が飛ぶようになります。
またコードを見て分かる通り、サーバシャットダウンを行う等の通知以外の処理も行うことができます。