assets_guardian.core.cache.cache

Classes

CacheManager([config])	Temporary cache and JSONL persistence manager.
LazyCacheIterable(service, file_path, item_type)	Iterable that reloads data from the disk at each iteration.

class assets_guardian.core.cache.cache.CacheManager(config: CacheConfig | None = None)

Bases: object

Temporary cache and JSONL persistence manager.

Allows offloading RAM by writing collected data to disk and managing checkpoints for crash recovery.

cleanup(command: str, env: AppEnv) → None

Deletes all temporary files associated with a given command.

Parameters:

command – Name of the command concerned.

get_file_path(command: str, source: str, instance_id: str, data_type: str) → Path

Generates a unique, command-isolated file path.

Parameters:

• command – Name of the command (e.g., ‘sync’, ‘audit’).

• source – Source name (e.g., ‘gitlab’).

• instance_id – Instance ID (e.g., ‘prod’).

• data_type – Data type (e.g., ‘identities’, ‘assets’, ‘accesses’).

has_checkpoint(path: Path) → bool

Checks if a checkpoint file exists.

Parameters:

path – Path of the checkpoint file.

load(file_path: Path, item_type: type[ModelType]) → Iterator

Loads a JSONL file as an iterator of typed objects by reading in blocks.

Reads the file in batches of N lines to optimize performance without saturating RAM.

Parameters:

• file_path – Path of the JSONL file to read.

• item_type – The class (dataclass) to deserialize into.

Yields:

ModelType – The deserialized objects one by one.

load_iterable(file_path: Path, item_type: type[ModelType]) → LazyCacheIterable

Returns an iterable object that reloads from disk without loading into RAM.

Parameters:

• file_path – Path of the JSONL file.

• item_type – The class to deserialize into.

Returns:

A reusable iterable tied to the file.

Return type:

LazyCacheIterable

save(items: Iterable[Any], file_path: Path) → None

Saves a list or an iterator of objects atomically.

Parameters:

• items – The objects to persist (can be a generator).

• file_path – Target path for the JSONL file.

class assets_guardian.core.cache.cache.LazyCacheIterable(service: CacheManager, file_path: Path, item_type: type[ModelType])

Bases: Generic

Iterable that reloads data from the disk at each iteration.