src/mcss/file_lock.py

"""Nonblocking advisory process locks for dataset maintenance scripts.

Windows retains the original one-byte msvcrt lock. POSIX uses an exclusive
flock on the open file description. Closing the handle releases either lock;
lock files may remain on disk and can safely be reused after a process exits.
"""

from __future__ import annotations

import os
from typing import BinaryIO


def acquire_file_lock(handle: BinaryIO) -> None:
    """Lock an open, nonempty file without waiting; contention raises OSError."""
    if os.name == "nt":
        import msvcrt

        handle.seek(0)
        msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
    else:
        import fcntl

        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)


def release_file_lock(handle: BinaryIO) -> None:
    """Release a previously acquired file lock without deleting its file."""
    if os.name == "nt":
        import msvcrt

        handle.seek(0)
        msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
    else:
        import fcntl

        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)