---
slug: "Djangoマネジメントコマンドの重複起動防止デコレータ"
title: "Django マネジメントコマンドの重複起動防止デコレータ"
description: ""
url: "https://www.ytyng.com/blog/Djangoマネジメントコマンドの重複起動防止デコレータ"
publish_date: "2015-11-27T00:40:36Z"
created: "2015-11-27T00:40:36Z"
updated: "2026-02-27T10:43:26.069Z"
categories: ["Django"]
keywords: ""
featured_image_url: "https://media.ytyng.com/resize/20230812/58b8b26bb6c4440ab71ff39a7f9b3e94.png.webp?width=768"
has_video: false
has_music: false
video_urls: []
music_urls: []
lang: "ja"
---

# Django マネジメントコマンドの重複起動防止デコレータ

<div class="document">


<pre class="literal-block">def find_process_ids(pattern):
    """
    pattern にマッチするプロセスIDを探して返す (自分以外)
    :rtype: list
    """
    self_pid = os.getpid()
    try:
        out = subprocess.check_output(['pgrep', '-f', pattern])
        return list(filter(
            lambda x: x != self_pid, map(int, out.splitlines())))
    except subprocess.CalledProcessError:
        return []


def runs_once(batch_name):
    """
    マネジメントコマンドの重複起動防止

    @runs_once(__file__)
    def handle(...):
        ...

    """
    re_batch_name = re.compile(r'^/.*/(\w+)\.py$')
    m = re_batch_name.match(batch_name)
    if m:
        batch_name = m.group(1)

    def _inner(func):
        @wraps(func)
        def decorate(*args, **kwargs):
            process_ids = find_process_ids(batch_name)
            if process_ids:
                print('process already exists. {}, {}'.format(
                    batch_name, process_ids)
                )
                return
            else:
                return func(*args, **kwargs)

        return decorate

    return _inner
</pre>
</div>