file_lock

utils.file_lock

File locking utilities for concurrent build safety.

Provides cross-platform file locking to prevent cache corruption when multiple build processes run simultaneously.

On Unix/macOS: Uses fcntl.flock() On Windows: Uses msvcrt.locking()

Example:

>>> from bengal.utils.file_lock import file_lock
>>> with file_lock(cache_path, exclusive=True):
...     # Safely read/write cache
...     cache.save(cache_path)

Classes

LockAcquisitionError

Raised when a lock cannot be acquired within the timeout.

Inherits from Exception

Functions

file_lock

Context manager for file locking. Acquires a lock on a .lock file adjacent to the target path. Use…

3 Generator[None]

def file_lock(path: Path, exclusive: bool = True, timeout: float = DEFAULT_LOCK_TIMEOUT) -> Generator[None]

Context manager for file locking.

Acquires a lock on a .lock file adjacent to the target path. Uses non-blocking attempts with retry for better timeout control.

Parameters 3

Name	Type	Default	Description
`path`	`Path`	—	Path to the file to lock (lock file will be path.lock)
`exclusive`	`bool`	`True`	If True, acquire exclusive (write) lock; else shared (read) lock
`timeout`	`float`	`DEFAULT_LOCK_TIMEOUT`	Maximum seconds to wait for lock (default: 30)

Returns

Generator[None]

_acquire_lock

Acquire a lock with timeout using non-blocking retries.

4 None

def _acquire_lock(lock_file: IO[str], exclusive: bool, timeout: float, lock_path: Path) -> None

Acquire a lock with timeout using non-blocking retries.

Parameters 4

Name	Type	Default	Description
`lock_file`	`IO[str]`	—	Open file handle for the lock file
`exclusive`	`bool`	—	If True, acquire exclusive lock; else shared lock
`timeout`	`float`	—	Maximum seconds to wait
`lock_path`	`Path`	—	Path to lock file (for logging)

_try_lock_nonblocking

Try to acquire lock without blocking.

2 None

def _try_lock_nonblocking(lock_file: IO[str], exclusive: bool) -> None

Try to acquire lock without blocking.

Parameters 2

Name	Type	Default	Description
`lock_file`	`IO[str]`	—	Open file handle
`exclusive`	`bool`	—	If True, acquire exclusive lock

_lock_unix

Acquire lock on Unix/macOS using fcntl.flock().

3 None

def _lock_unix(lock_file: IO[str], exclusive: bool, blocking: bool) -> None

Acquire lock on Unix/macOS using fcntl.flock().

Parameters 3

Name	Type	Default	Description
`lock_file`	`IO[str]`	—	Open file handle
`exclusive`	`bool`	—	If True, acquire exclusive (LOCK_EX); else shared (LOCK_SH)
`blocking`	`bool`	—	If False, raise BlockingIOError if lock unavailable

_lock_windows

Acquire lock on Windows using msvcrt.locking().

3 None

def _lock_windows(lock_file: IO[str], exclusive: bool, blocking: bool) -> None

Acquire lock on Windows using msvcrt.locking().

Parameters 3

Name	Type	Default	Description
`lock_file`	`IO[str]`	—	Open file handle
`exclusive`	`bool`	—	If True, acquire exclusive lock (always exclusive on Windows)
`blocking`	`bool`	—	If False, raise BlockingIOError if lock unavailable

_release_lock

Release lock on the file.

1 None

def _release_lock(lock_file: IO[str]) -> None

Release lock on the file.

Parameters 1

Name	Type	Default	Description
`lock_file`	`IO[str]`	—	Open file handle with lock

_unlock_unix

Release Unix/macOS lock.

1 None

def _unlock_unix(lock_file: IO[str]) -> None

Release Unix/macOS lock.

Parameters 1

Name	Type	Default	Description
`lock_file`	`IO[str]`	—

_unlock_windows

Release Windows lock.

1 None

def _unlock_windows(lock_file: IO[str]) -> None

Release Windows lock.

Parameters 1

Name	Type	Default	Description
`lock_file`	`IO[str]`	—

is_locked

Check if a file is currently locked.

1 bool

def is_locked(path: Path) -> bool

Check if a file is currently locked.

Parameters 1

Name	Type	Default	Description
`path`	`Path`	—	Path to check

Returns

bool —

True if file appears to be locked by another process

remove_stale_lock

Remove a stale lock file that may have been left by a crashed process.

2 bool

def remove_stale_lock(path: Path, max_age_seconds: float = 3600) -> bool

Remove a stale lock file that may have been left by a crashed process.

Parameters 2

Name	Type	Default	Description
`path`	`Path`	—	Path to the file (not the lock file)
`max_age_seconds`	`float`	`3600`	Maximum age in seconds before considering stale

Returns

bool —

True if stale lock was removed