Djangoカスタムコマンドの二重起動の防止

INDEX

はじめに
環境
テスト
ソース
参考

はじめに

Djangoのカスタムコマンドを作成してcronで1分毎に動作させる案件がありました。
ただ、処理するデータ量によっては1分以内に処理が完了しないこともあるので、二重起動を防止しようと考えました。
以下のサイトに情報があったのですが、Python3でcommandsモジュールは削除されたので、subprocessを使うように書き換えました。

djangoコマンドの二重起動を防止する - Qiita

djangoコマンドとは'python manage.py 〜'で実行するコマンドのことです。デフォルトでshellなどたくさん便利なコマンドが用意されています。ここでは詳しく説明しませんがカス…

pythonでプロセスの多重起動を防ぐ - Qiita

python3系でプロセスの2重起動とかの多重起動を防ぐ方法です。多重起動を防ぐ一般的な方法としては大体以下の２つかと。ロックファイルを作るプロセスの状況をOSなどから監視するロックファイル…

他には以下のサイトもありましたが、よく理解できなかったので上記を使用しました。
　https://gist.github.com/ytyng/385b6dbb216d77a1d8ed

環境

Debian bullseye 64bit
Python 3.10.2
Djnago 4.0.2

テスト

ターミナルで以下のコマンドを実行するとtime.sleep(60)で60秒間待機します。
異なるターミナルで同コマンドを実行しても二重起動はできません。

python manage.py hoge

ソース

 
import subprocess
import sys
from django.core.management.base import BaseCommand, CommandError

class Command(BaseCommand):
    # 「python manage.py help hoge」で表示されるメッセージ
    help = 'Hoge Script'

    #
    # プロセスが動いていたらTrueを返す
    #
    def is_process_exist(self, process_name):
        p1 = subprocess.Popen(['ps', '-ef'], stdout=subprocess.PIPE)
        p2 = subprocess.Popen(['/usr/bin/grep', process_name], stdin=p1.stdout, stdout=subprocess.PIPE)
        p3 = subprocess.Popen(['/usr/bin/grep', '-v', 'grep'], stdin=p2.stdout, stdout=subprocess.PIPE)
        p4 = subprocess.Popen(['/usr/bin/wc', '-l'], stdin=p3.stdout, stdout=subprocess.PIPE)
        p1.stdout.close()
        p2.stdout.close()
        p3.stdout.close()
       #output = p4.communicate()[0]
        output   = p4.communicate()[0].decode("utf8").replace('\n','')
        return int(output) >= 2

    # コマンドが実行された際に呼ばれるメソッド
    def handle(self, *args, **options):
        # 二重起動チェック
        if self.is_process_exist(__name__.split('.')[-1]):
            sys.exit('No double activation.')

        # 二重起動のテスト用
        import time
        time.sleep(60)

        ・・・・・
        ・・・・・