pythonの簡単なファイルロック

check.bash

for i in `seq 0 4`; do python check.py & done
# [1] 83406
# [2] 83407
# [3] 83408
# [4] 83409
# [5] 83410

# start sleep 0
# finish sleep 0
# start sleep 1
# finish sleep 1
# start sleep 3
# finish sleep 3
# start sleep 6
# finish sleep 6
# start sleep 9
# ...

filelock.py

import fcntl
from time import sleep

import os
from multiprocessing.pool import Pool
from pathlib import Path


class FileLock(object):
    def __init__(self, lock_file: Path):
        self.lock_file = lock_file
        self._lock_file_fd = None

    def _acquire(self):
        open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC
        fd = os.open(str(self.lock_file), open_mode)

        try:
            fcntl.flock(fd, fcntl.LOCK_EX)
        except (IOError, OSError):
            os.close(fd)
        else:
            self._lock_file_fd = fd

    def _release(self):
        fd = self._lock_file_fd
        self._lock_file_fd = None
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)

    def __enter__(self):
        self._acquire()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._release()


def process(t: int):
    lock = FileLock(lock_file=Path('hoge.lock'))
    with lock:
        print('start sleep', t)
        sleep(t)
        print('finish sleep', t)


def main():
    it = Pool().imap_unordered(process, range(10))
    list(it)


if __name__ == '__main__':
    main()