torchcell.utils.FileLockHelper

class FileLockHelper[source]

Bases: object

Cross-platform file locking utility for JSON file operations.

This class provides thread-safe and process-safe file operations using the filelock library, which works across all platforms including Windows, macOS, and Linux.

default_timeout

Default timeout in seconds for acquiring locks. A caller’s timeout replaces it only when given (None means the default), so timeout=0 is one non-blocking attempt, as in filelock.

Type:

float

default_retry_delay

Default delay between retries in seconds, passed to filelock’s acquire as poll_interval.

Type:

float

classmethod read_json_with_lock(file_path: str | Path, timeout: float | None = None, retry_delay: float | None = None, create_if_missing: bool = False, default_data: Any | None = None) → Any[source]

Read JSON file with file locking for thread/process safety.

Parameters:

  • file_path – Path to the JSON file

  • timeout – Maximum time to wait for lock acquisition (None: the class default; 0: a single non-blocking attempt)

  • retry_delay – Delay between lock acquisition retries (None: the class default)

  • create_if_missing – If True, create file with default_data if it doesn’t exist

  • default_data – Data to write if creating new file (defaults to empty dict)

Returns:

Parsed JSON data

Raises:

  • Timeout – If lock cannot be acquired within timeout

  • json.JSONDecodeError – If file contains invalid JSON

  • FileNotFoundError – If file doesn’t exist and create_if_missing is False

classmethod write_json_with_lock(file_path: str | Path, data: Any, timeout: float | None = None, retry_delay: float | None = None, indent: int = 2, ensure_ascii: bool = False) → None[source]

Write JSON file with file locking for thread/process safety.

Parameters:

  • file_path – Path to the JSON file

  • data – Data to serialize to JSON

  • timeout – Maximum time to wait for lock acquisition (None: the class default; 0: a single non-blocking attempt)

  • retry_delay – Delay between lock acquisition retries (None: the class default)

  • indent – JSON indentation level

  • ensure_ascii – If True, escape non-ASCII characters

Raises:

  • Timeout – If lock cannot be acquired within timeout

  • TypeError – If data cannot be serialized to JSON

classmethod update_json_with_lock(file_path: str | Path, update_func: Callable[[Any], Any], timeout: float | None = None, retry_delay: float | None = None, create_if_missing: bool = True, default_data: Any | None = None) → Any[source]

Update JSON file atomically with a custom function.

This method reads the file, applies the update function, and writes the result back, all within a single lock acquisition.

Parameters:

  • file_path – Path to the JSON file

  • update_func – Function that takes current data and returns updated data

  • timeout – Maximum time to wait for lock acquisition (None: the class default; 0: a single non-blocking attempt)

  • retry_delay – Delay between lock acquisition retries (None: the class default)

  • create_if_missing – If True, create file with default_data if it doesn’t exist

  • default_data – Initial data if creating new file

Returns:

The updated data

Raises:

Timeout – If lock cannot be acquired within timeout

classmethod with_file_lock(file_path: str | Path, timeout: float | None = None) → UnixFileLock[source]

Get a FileLock instance for manual lock management.

This is useful when you need more control over the locking process or need to perform non-JSON operations.

Parameters:

  • file_path – Path to the file to lock

  • timeout – Maximum time to wait for lock acquisition (None: the class default; 0: a single non-blocking attempt)

Returns:

FileLock instance

Example:

with FileLockHelper.with_file_lock("data.txt") as lock:
    # Perform thread-safe operations
    with open("data.txt", 'a') as f:
        f.write("new data\\n")
classmethod cleanup_lock_files(directory: str | Path) → int[source]

Remove all lock files in a directory.

This should only be called when you’re certain no processes are using the locks (e.g., during cleanup or testing).

Parameters:

directory – Directory to clean up

Returns:

Number of lock files removed