assets_guardian.core.domain.engines.collector_engine

Classes

CollectorEngine([cache])	IAM collection engine with checkpoint (caching) support.
CollectorResult(source_name, instance_id, ...)	Result of a single collector execution.

class assets_guardian.core.domain.engines.collector_engine.CollectorEngine(cache: CacheManager | None = None)

Bases: object

IAM collection engine with checkpoint (caching) support.

run_collect(collector: Any, mode: AssetsGuardianMode) → CollectorResult

Runs a collector with checkpoint (caching) support.

For each data type, checks if it is already present in the cache; if not, triggers a live collection.

Parameters:

• collector – The collector instance to execute.

• mode – The command mode (‘sync’ or ‘audit’) used to determine the cache namespace.

Returns:

The result containing collected data or failure info.

Return type:

CollectorResult

class assets_guardian.core.domain.engines.collector_engine.CollectorResult(source_name: str, instance_id: str, identities: Iterable[Identity] = <factory>, assets: Iterable[Asset] = <factory>, accesses: Iterable[Access] = <factory>, success: bool = True, error: str | None = None)

Bases: object

Result of a single collector execution.

Variables:

• source_name (str) – Name of the source (e.g., ‘gitlab’).

• instance_id (str) – ID of the collection instance (e.g., ‘prod’).

• identities (collections.abc.Iterable[assets_guardian.core.domain.models.identity.Identity]) – Iterable of collected identities.

• assets (collections.abc.Iterable[assets_guardian.core.domain.models.asset.Asset]) – Iterable of collected assets.

• accesses (collections.abc.Iterable[assets_guardian.core.domain.models.access.Access]) – Iterable of collected access data.

• success (bool) – Flag indicating if the collection succeeded.

• error (str | None) – Error message if the collection failed.