[PYTHON] Django Management Command Duplicate Launch Decorator
decorators.py
def find_process_ids(pattern):
"""
Finds and returns the process ID that matches the pattern(Other than your own)
: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):
"""
Preventing duplicate activation of management commands
@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