key_lock

class KeyLock:
    """A handle on the lock of ``key``: ``acquire`` or ``try_acquire`` it, ``release`` it."""

    def __init__(self, directory: Path, key: str) -> None:
        """Open the lock file of ``key`` under ``directory`` without acquiring it.

        Args:
            directory: Where the lock files of this kind of key live.
            key: The key to hold.
        """
        self.key = key
        self.path = directory / lock_file_name(key)
        self._handle: IO[bytes] | None = None
        self._held = False
        self._handle = open(self.path, "ab")

    def acquire(self, timeout_seconds: float | None) -> Self:
        """Hold the lock, waiting for another holder up to ``timeout_seconds``.

        Args:
            timeout_seconds: How long to wait; ``None`` waits as long as it takes.

        Raises:
            RecipeError: If the lock is still held elsewhere when the timeout
                passes, or this handle has been released.
        """
        handle = self._open_handle()
        if timeout_seconds is None:
            fcntl.flock(handle, fcntl.LOCK_EX)
            self._held = True
            return self
        deadline = time.monotonic() + max(timeout_seconds, 0.0)
        while not self.try_acquire():
            if time.monotonic() >= deadline:
                self.release()
                raise RecipeError(f"lock on {self.key!r} not acquired within {timeout_seconds}s")
            time.sleep(RETRY_SECONDS)
        return self

    def try_acquire(self) -> bool:
        """Hold the lock if nobody else does, without waiting.

        Raises:
            RecipeError: If this handle has been released.
        """
        try:
            fcntl.flock(self._open_handle(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return False
        self._held = True
        return True

    def claim(self, note: str) -> None:
        """Write ``note`` as the lock file's content, replacing what a past holder left.

        Raises:
            RecipeError: If this lock is not held.
        """
        if not self.held:
            raise RecipeError(f"lock on {self.key!r} is not held; nothing to claim")
        handle = self._open_handle()
        handle.seek(0)
        handle.truncate()
        handle.write(note.encode())
        handle.flush()

    def claimant(self) -> str:
        """The note the current or last holder wrote, or ``""``."""
        try:
            return self.path.read_text()
        except FileNotFoundError:
            return ""

    @property
    def held(self) -> bool:
        """Whether this handle holds the lock."""
        return self._held

    def release(self) -> None:
        """Let the next holder in and close the handle."""
        if self._handle is not None:
            self._handle.close()
            self._handle = None
        self._held = False

    def __enter__(self) -> Self:
        """Hold the lock for the block."""
        return self

    def __exit__(self, *_: object) -> None:
        """Release the lock."""
        self.release()

    def __del__(self) -> None:
        """Release the lock when the handle is collected."""
        self.release()

    def _open_handle(self) -> IO[bytes]:
        if self._handle is None:
            raise RecipeError(f"lock on {self.key!r} has been released")
        return self._handle