Python file locking using `fcntl.flock`

flock_example.py

#! /usr/bin/env python
"""
Example of using fcntl.flock for locking file. Some code inspired by filelock module.
"""

import fcntl
import os
import time


def acquire(lock_file) -> None | int:
    open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC
    fd = os.open(lock_file, open_mode)

    pid = os.getpid()
    lock_file_fd = None

    timeout = 60
    start_time = current_time = time.time()
    while current_time < start_time + timeout:
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (IOError, OSError):
            pass
        else:
            lock_file_fd = fd
            print(f"  {pid} acquired lock")
            break
        print(f"  {pid} waiting for lock")
        time.sleep(1.0)
        current_time = time.time()
    if time.time() > start_time + timeout:
        print(f"  {pid} timed out waiting for lock")
    if lock_file_fd is None:
        os.close(fd)
    return lock_file_fd


def release(lock_file_fd) -> None:
    # Do not remove the lockfile:
    #
    #   https://github.com/benediktschmitt/py-filelock/issues/31
    #   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
    fcntl.flock(lock_file_fd, fcntl.LOCK_UN)
    os.close(lock_file_fd)
    print(f"  {os.getpid()} released lock")
    return None


def is_locked(lock_file: str) -> bool:
    """Checks if the lock file is currently locked by another process.
    Returns True if locked, False otherwise.
    """
    open_mode = os.O_RDWR | os.O_CREAT
    fd = os.open(lock_file, open_mode)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (IOError, OSError):
        os.close(fd)
        return True
    else:
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)
        return False


def this_may_raise():
    raise ValueError("Simulate exception in critical section")


def main() -> None:
    pid = os.getpid()
    print(f"{pid} started")
    fd = acquire("myfile.lock")
    assert is_locked("myfile.lock"), f"{pid} lock NOT acquired"

    try:
        time.sleep(5.0)
        this_may_raise()
        # the release never runs because of the exception raised
        release(fd)
    except Exception:
        pass
    # finally:
    #     # this release always runs
    #     release(fd)

    if is_locked("myfile.lock"):
        print(f"{pid} lock NOT released;")
        # if this program takes long to finish, other
        # processes will have to wait its entire duration.
    else:
        print(f"{pid} lock released")

    time.sleep(10)  # simulate a long-running process
    print(f"{pid} finished")


if __name__ == "__main__":
    main()