locking

flavor.locking

TODO: Add module docstring.

Classes

LockError

Bases: Exception

Error during lock operations.

LockManager

LockManager(lock_dir: Path | None = None)

Manages file-based locks for concurrent operations.

Source code in flavor/locking.py

def __init__(self, lock_dir: Path | None = None) -> None:
    self.lock_dir = lock_dir or Path.home() / ".cache" / "flavor" / "locks"
    ensure_dir(self.lock_dir)
    self.held_locks: set[Path] = set()

Functions

cleanup_all

cleanup_all() -> None

Clean up all held locks (for emergency cleanup).

Source code in flavor/locking.py

def cleanup_all(self) -> None:
    """Clean up all held locks (for emergency cleanup)."""
    # Foundation FileLock handles cleanup automatically
    self.held_locks.clear()

lock

lock(
    name: str, timeout: float = 30.0
) -> Generator[Path, None, None]

Acquire a named lock.

Parameters:

Name	Type	Description	Default
name	str	Lock name	required
timeout	float	Maximum time to wait for lock	30.0

Yields:

Type	Description
Path	Lock file path

Raises:

Type	Description
LockError	When unable to acquire lock

Source code in flavor/locking.py

@contextmanager
def lock(self, name: str, timeout: float = 30.0) -> Generator[Path, None, None]:
    """
    Acquire a named lock.

    Args:
        name: Lock name
        timeout: Maximum time to wait for lock

    Yields:
        Lock file path

    Raises:
        LockError: When unable to acquire lock
    """
    lock_file = self.lock_dir / f"{name}.lock"

    file_lock = FileLock(lock_file, timeout=timeout)
    try:
        acquired = file_lock.acquire(blocking=True)
        if not acquired:
            raise LockError(f"Timeout acquiring lock: {name}")

        self.held_locks.add(lock_file)
        try:
            yield lock_file
        finally:
            file_lock.release()
            self.held_locks.discard(lock_file)
    except Exception as e:
        if "timeout" in str(e).lower():
            raise LockError(f"Timeout acquiring lock: {name}") from e
        raise