delhi09の勉強日記

技術トピック専用のブログです。自分用のメモ書きの投稿が多いです。あくまで「勉強日記」なので記事の内容は鵜呑みにしないでください。

Djangoのカスタムコマンドの実装をチームで最低限統一する方法の検討

Django

はじめに

この記事はDjango Advent Calendar 2023の10日目の記事です。

Djangoでバッチ処理を作る際には、カスタムコマンドを実装することが多いと思います。

カスタムコマンドで実装されたバッチ処理において、以下のような考慮の有無が実装担当者に依存していたり、されている場合でも実装方法は統一されていなかったりするということを業務で経験しました。

  • 開始と終了のログの出力
  • 想定外エラーのハンドリング
  • 多重起動の防止

そこで、本記事ではチームで最低限統一するために開発開始時点で打てる施策を考えてみたいと思います。

問題点

その前に先にあげたような事項が考慮されないと何が問題かをもう少し説明します。

開始と終了のログの出力

開始と終了のログがないと、そもそもバッチが実行されたかどうかが分かりません。これがないと最低限の監視ができないですし、バッチの実行時間を知りたい時などにも困ります。

想定外エラーのハンドリング

もちろん想定できる例外は個別にハンドリングした方がいい場合が多いですが、ネットワークエラーや考慮漏れによる想定外のエラーが発生することも想定されます。その場合に、処理がabortするのはよくありません。

多重起動の防止

バッチをcronなどで定期実行している場合、何らかの理由で実行時間が想定をオーバーして、前回のバッチが終了していないのに次回のバッチが起動してしまうということもありえます。バッチの性質によっては障害に繋がります。

実装担当者が「このバッチは多重起動してもOK」と分かった上で、チェックしていないのであればよいのですが、そもそもそういう処理が必要な場合があるという知識がなかったり、考慮を忘れていたりするのであれば問題です。

結論

結論としては、以下のようなベースコマンドを実装しておくとよいのではないかと思っています。

from django.core.management.base import BaseCommand

import logging
import sys
logger = logging.getLogger(__name__)


class BaseProjectCommand(BaseCommand):
    def command_name(self)->str:
        raise NotImplementedError("コマンド名を記述してください。")
    
    def start_log(self):
        raise NotImplementedError("開始ログは必ず出力してください。 (ex. hoge_command start.)")
    
    def end_log(self):
        raise NotImplementedError("終了ログは必ず出力してください。 (ex. hoge_command end.)")
    
    def process(self, *args, **options):
        raise NotImplementedError("ビジネスロジックを記述してください。")
    
    def check_is_already_running(self)->bool:
        raise NotImplementedError("多重起動チェック処理を記述してください。不要な場合はFalseを返してください。")

    def handle(self, *args, **options):
        if self.check_is_already_running():
            logger.warning("プロセスが既に存在するので起動しません。[name=%s]", self.command_name())
        self.start_log()
        try:
            self.process(*args, **options)
        except Exception:
            logger.exception("%s end with error.", self.command_name())
            sys.exit(1)
        self.end_log()

加えてチームのコーディングルールとして以下を定めます。

  1. カスタムコマンドは必ずこのベースコマンドを継承して実装する
  2. ビジネスロジックはprocessメソッドに実装する
  3. 想定外エラーのハンドリングは個別のバッチでは実装しない

継承先は例えば以下のように実装します。

from .base_project_command import BaseProjectCommand

import logging
logger = logging.getLogger(__name__)


class Command(BaseProjectCommand):
    def command_name(self) -> str:
        return "test_batch"
    
    def start_log(self):
        logger.info("%s start.", self.command_name())
    
    def end_log(self):
        logger.info("%s end.", self.command_name())

    def check_is_already_running(self) -> bool:
        return False
    
    def process(self, *args, **options):
        print("test")

説明

command_name

親クラス側でコマンド名を取得するために実装しています。

開始ログ/終了ログ

親クラス側でそれぞれstart_log、end_logというメソッドを定義して、NotImplementedErrorを送出することで実装が漏れていたら気づけるようにしています。

command_nameを実装しているので、ログ出力まで親クラス側で実装してしまうこともできますが、以下の理由より実装は個別のバッチに任せた方が任せた方がいいケースが多いのかなと思っています。

  • 開始ログ: オプションなども出力したい場合がある
  • 終了ログ: 成功件数や失敗件数なども出力したい場合がある

想定外エラーのハンドリング

バッチ個別のビジネスロジックはprocessメソッドに実装してもらうことで、想定外のエラーのハンドリングを親クラスで共通化しています。これによって以下のようなメリットがあると考えられます。

  • 実装担当者によって想定外エラーのハンドリングがあったりなかったりするのを防げる & やり方も統一化できる
  • バッチ実装の度にエラーハンドリングするコストを減らせる(若干ですが)

多重起動の防止

親クラス側でcheck_is_already_runningというメソッドを定義して、NotImplementedErrorを送出しています。もちろん、多重起動しても問題ないバッチも多いので、その場合はFalseを返せばいいようにしています。

こうやっておくことで、「多重起動しても問題ないバッチか？」ということを考慮したりチーム内で議論するきっかけになったりする効果があると思っています。

多重起動の防止の具体的な方法については、各々のバッチの実行環境(1つのバッチサーバで動いている or スケールさせているなど)やどの程度厳しく防ぎたいかによると思うので、ここでは触れません。私は簡単にgrepでチェックする方法や、DBにステータスを持って管理する方法の経験があります。

参考

以下の記事はこの記事よりも大分レベルが高い話ですが、バッチ設計時の考慮事項がだいたい書かれていて勉強になります。 engineering.mercari.com

追記

同僚からフィードバックを頂きました。

サンプルコードでsys.exit(1)しているところは、CommandErrorをラップして送出するのでもいいそうです。 たしかに、公式ドキュメントを読むと内部でsys.exit()しておりデフォルトのリターンコードは1であると書かれています。

from django.core.management.base import CommandError

# ...

except Exception as e:
        logger.exception("%s end with error.", self.command_name())
        raise CommandError from e