CloudGEngine

The engine behind cloudg run, cloudg ingest and cloudg map: call the whole pipeline in one line, or drive each phase yourself.

Which method to call

CloudGEngine takes one argument, a CloudGConfig, and keeps it on engine.config. Everything else is a method call. The methods fall into three groups, and they share very little state: the engine holds the config and the six event hooks, nothing more. You can call collect() twice, or call analyze() on assets you built by hand, and nothing on the engine gets in the way.

flowchart LR
  RP["run_pipeline()"] --> C["collect()"]
  RP --> S["scan()"]
  RP --> A["analyze()"]
  RP --> N["FindingsNormaliser"]
  RP --> R["findings.json, report.html, topology.svg"]
  RFR["run_from_reports()"] --> I["ingest_reports()"]
  RFR --> NF["normalise_findings()"]
  NF --> N
  RFR --> R
  MI["map_inventory()"] --> IM["InventoryMapper"]

What each entry point runs

You have	Call	Cloud access	Scanner binaries
Credentials, and want everything cloudg run produces	run_pipeline()	yes	the ones in scanners.enabled
Prowler, ScoutSuite, Checkov or Trivy output from somewhere else	run_from_reports()	no	no
Credentials, and only want to know what exists and how it connects	map_inventory()	yes	no
Your own assets, or a need to stop between phases	collect(), scan(), analyze(), normalise_findings()	for collect() and the cloud scanners	for scan()

The *_sync variants wrap the async method in asyncio.run(). Use them from plain scripts. Inside a running event loop (a Jupyter notebook, say) they raise RuntimeError with a message naming the async method to await instead.

output_dir is optional on scan(), analyze(), run_from_reports(), run_pipeline() and their sync wrappers. Left out (or None), it is report.output_dir from the config, ./reports by default.

What run_pipeline does, phase by phase

run_pipeline(output_dir=None) creates the output directory, then:

1. collect() runs the collectors for every provider in config.providers, across aws.regions, aws.accounts, the Azure subscriptions and the GCP projects you configured.
2. scan() builds a graph of the collected assets and runs the reachability analysis on it, then starts every scanner in scanners.enabled in a thread pool with one worker per job: Prowler and ScoutSuite once per provider, with aws.profile and the rest of the configured AWS credentials and regions; Checkov once per IaC directory; Trivy once over scanners.trivy_images, or over the IaC directories when there are no images; scanner plugins by name. The IAM policy linter runs when iam is in the list and there are assets. Unknown names are logged and skipped.
3. analyze() rebuilds the graph and records node and edge counts, reachability findings and lateral movement paths. It then writes the ontology (ontology.enabled), the RAG chunks (rag.enabled) and the Terraform recreation (terraform.enabled).
4. The scanner findings and the reachability findings go through FindingsNormaliser: deduplication, cross-scanner merging, CVSS rescoring and compliance mapping. The reachability analysis ran in both step 2 and step 3. Both runs produce the same check on the same resource, so the deduplication pass folds the copies into one.
5. The reports in report.formats are written: findings.json and report.html with the D3 graph embedded, and topology.svg.

Everything ends up in one PipelineResult.

IaC targets for Checkov

scan() picks the directories Checkov scans in this order: the iac_dir argument, then scanners.iac_directories, then a Terraform recreation of the live assets (generated on the spot when terraform.enabled is true). When none of these exists Checkov is skipped with a warning. cloudg no longer falls back to the current directory, because scanning whatever happens to be in your working directory says nothing about the cloud. The same logic is public as cloudg.api.resolve_iac_dirs(iac_dir, config_dirs, terraform_dir=None), which returns the directories together with their source: "cli", "config", "terraform", or None when nothing qualifies.

Examples

Full pipeline

This needs read access to the account (any method from Authentication works) and the scanner binaries you enable. Missing binaries are skipped.

nightly_pipeline.py

import asyncio
import json

from cloudg import CloudGConfig, CloudGEngine

async def main() -> None:
    config = CloudGConfig(providers=["aws"])
    config.aws.profile = "security-audit"
    config.aws.regions = ["eu-west-1", "eu-central-1"]
    config.scanners.enabled = ["prowler", "trivy"]
    config.scanners.trivy_images = [
        "123456789012.dkr.ecr.eu-west-1.amazonaws.com/orders-worker:latest"
    ]

    engine = CloudGEngine(config)
    result = await engine.run_pipeline(output_dir="./reports")

    print(json.dumps(result.to_summary(), indent=2))
    print(result.report_paths)  # {'json': ..., 'html': ..., 'svg': ...}

    # Collection failures don't raise; they are recorded per service
    for cov in result.coverage:
        if cov.failed_services:
            print("incomplete:", cov.to_summary())

asyncio.run(main())

Ingest only

No credentials and no scanners: point the engine at reports produced elsewhere. Keys are tool names (prowler, scoutsuite, checkov, trivy); values are lists of files or directories.

ingest_only.py

from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig())
engine.on_phase_start = lambda phase: print("phase:", phase)

result = engine.run_from_reports_sync(
    {
        "prowler": ["./prowler-output/"],
        "checkov": ["./results_json.json"],
        "trivy": ["./trivy.json"],
    },
    output_dir="./reports",
)

print(result.total_findings, result.severity_breakdown)
print(result.report_paths)
for f in result.findings:
    print(f.severity.value, f.risk_score, f.source_tool, f.title)

With one failing Prowler check, one Checkov failure and one Trivy CVE, it prints:

$ python ingest_only.py
phase: ingest
phase: normalisation
phase: reporting
3 {'CRITICAL': 1, 'HIGH': 2}
{'json': PosixPath('reports/findings.json'), 'html': PosixPath('reports/report.html')}
CRITICAL 9.7 trivy [Trivy] CVE-2024-0001: openssl (alpine)
HIGH 7.5 prowler S3 bucket default encryption
HIGH 7.5 checkov [Checkov/terraform] Ensure S3 bucket has server-side encryption enabled

The result has no assets, so graph_nodes, ontology_triples and attack_paths stay empty. To get them as well, collect live assets and pass the ingested findings to analyze() (next examples).

Map only

map_inventory() runs the deep inventory collectors and the relationship linker. No scanner is involved. Passing output_dir writes the map files; passing findings as well writes asset-map.json and compliance-map.json.

map_only.py

from cloudg import CloudGConfig, CloudGEngine

config = CloudGConfig(providers=["aws"])
config.aws.regions = ["ALL"]
config.inventory.exclude_services = ["kubernetes"]

engine = CloudGEngine(config)
inventory = engine.map_inventory_sync(output_dir="./inventory", tagging_sweep=False)

summary = inventory.summary
print(summary["total_assets"], "assets,", summary["total_edges"], "edges")
print("internet exposed:", summary["internet_exposed"])
print("top services:", list(summary["assets_by_service"].items())[:5])

The return value is an InventoryResult. Unlike collect(), this method re-raises when the mapping fails, after calling on_error("inventory_mapping", exc).

Phases by hand

Driving the phases yourself lets you swap one out. This script runs offline: two hand-built assets stand in for collect(), an existing Trivy report stands in for scan(), and the graph, ontology and RAG outputs are produced from them.

phases_offline.py

import asyncio

from cloudg import CloudGConfig, CloudGEngine
from cloudg.schema.models import AssetType, CloudAsset, CloudProvider, EdgeType, NetworkEdge

# Two assets and two edges, as a collector or a CMDB export would supply them
sg = CloudAsset(
    arn="arn:aws:ec2:eu-west-1:123456789012:security-group/sg-0a1b2c3d4e5f60718",
    name="bastion-sg",
    asset_type=AssetType.SECURITY_GROUP,
    provider=CloudProvider.AWS,
    region="eu-west-1",
    account_id="123456789012",
)
vm = CloudAsset(
    arn="arn:aws:ec2:eu-west-1:123456789012:instance/i-0b1c2d3e4f5061728",
    name="bastion",
    asset_type=AssetType.EC2,
    provider=CloudProvider.AWS,
    region="eu-west-1",
    account_id="123456789012",
)
edges = [
    NetworkEdge(
        source_id="0.0.0.0/0",
        target_id=sg.id,
        edge_type=EdgeType.SECURITY_GROUP_RULE,
        cidr="0.0.0.0/0",
        port_range="22",
        protocol="TCP",
    ),
    NetworkEdge(source_id=sg.id, target_id=vm.id, edge_type=EdgeType.ATTACHED_TO),
]

async def main() -> None:
    engine = CloudGEngine(CloudGConfig(providers=["aws"]))
    engine.on_phase_start = lambda phase: print("phase:", phase)

    findings = engine.ingest_reports({"trivy": ["./trivy.json"]})
    analysis = await engine.analyze([sg, vm], edges, findings, output_dir="./reports")
    scan_result = engine.normalise_findings(
        findings + analysis.reachability_findings, assets=[sg, vm]
    )

    print(analysis.graph_nodes, "nodes,", analysis.graph_edges, "edges")
    for f in analysis.reachability_findings:
        print(f.severity.value, f.title)
    print("ontology:", analysis.ontology_triples, "triples, last file", analysis.ontology_path)
    print("RAG chunks:", analysis.rag_chunks_path)
    print(scan_result.summary["severity_breakdown"])

asyncio.run(main())

$ python phases_offline.py
phase: ingest
phase: analysis
3 nodes, 2 edges
CRITICAL Security group allows SSH (port 22) from 0.0.0.0/0
ontology: 670 triples, last file reports/ontology.jsonld
RAG chunks: reports/rag_chunks.jsonl
{'CRITICAL': 2}

The graph has three nodes for two assets: 0.0.0.0/0 becomes a node of its own. With credentials, replace the hand-built lists with collection = await engine.collect() and pass collection.assets and collection.edges, and replace the ingest with await engine.scan(...) if you want the scanners to run.

Hooks

Hooks are plain attributes. Set them before calling a method; each one is optional.

hooks.py

from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig())
engine.on_phase_start = lambda phase: print(f"-> {phase}")
engine.on_finding = lambda f: print(f"   {f.severity.value:8} {f.title}")
engine.on_error = lambda phase, exc: print(f"!! {phase}: {exc}")

engine.run_from_reports_sync({"trivy": ["./trivy.json"]}, output_dir="./reports")

Event hooks lists every hook, when it fires and what it receives, with longer examples.

Notes

PipelineResult.errors is the run's full error list, one "<phase>: <message>" string per failure. It holds everything reported through on_error during the run (scanners, scanner credentials, graph, ontology, RAG, Terraform, normalisation, reporting) plus one collection: ... line per provider, account or region that could not be collected. An empty list means the run was complete. result.coverage still has the per-service detail.

collect() never raises for a collection problem. When the collector itself throws, the engine calls on_error("collection", exc) and returns an empty CollectionResult, and run_pipeline() carries on with no assets.

on_finding fires once per finding as each scanner finishes in scan() (and when ingest_reports() returns); on_scan_complete fires once at the end. Both run before normalisation, so the findings have not been deduplicated or rescored yet. They receive deep copies, which normalisation does not change later. For the final list, read result.findings after the run.

map_inventory(findings=...) stores the overlay on the returned result as result.asset_map and result.compliance_map, with or without output_dir. With output_dir set, it also writes asset-map.json and compliance-map.json there. To overlay findings on a map you already hold, call InventoryMapper.export_merged() yourself.

The Terraform recreation goes to <output_dir>/terraform, following the output_dir you pass to run_pipeline(), analyze() or scan(). A terraform.output_dir set in the config replaces it, except the old default ./reports/terraform, which counts as unset.

scanners.timeout_seconds (default 3600) is the subprocess timeout of each scanner job: each Prowler or ScoutSuite run, each Checkov directory, each Trivy image or directory. A job that runs over is killed and reported through on_error under its job name, and the findings it produced before that are kept. A scanner that fails to launch is reported the same way.

Reference

CloudGEngine

CloudGEngine(config: CloudGConfig)

Bases: ScannerRunsMixin

Integration-ready programmatic API for CloudG.

Provides fine-grained control over the pipeline for embedding in larger security orchestration systems.

Example

engine = CloudGEngine(config)
engine.on_finding = lambda f: send_to_siem(f)
result = await engine.run_pipeline()

Source code in cloudg/api.py

def __init__(self, config: CloudGConfig) -> None:
    self.config = config

    # Event hooks: set these before calling run_pipeline()
    self.on_collection_complete: OnCollectionComplete | None = None
    self.on_scan_complete: OnScanComplete | None = None
    self.on_finding: OnFinding | None = None
    self.on_analysis_complete: OnAnalysisComplete | None = None
    self.on_phase_start: OnPhaseStart | None = None
    self.on_error: OnError | None = None

    # While run_pipeline() runs, every emitted error is also kept here
    self._error_sink: list[str] | None = None

run_pipeline async

run_pipeline(output_dir: str | Path | None = None) -> PipelineResult

Run the complete CloudG pipeline.

Phases: 1. collection (multi-provider, multi-region), 2. scanning (reachability and every enabled scanner), 3. analysis (ontology, RAG, Terraform), 4. normalisation, 5. report generation (report.formats). Files go to output_dir, or report.output_dir when it is None.

Returns:

Type	Description
PipelineResult	PipelineResult with all outputs. errors lists every failure of the run, each as "<phase>: <message>": everything reported through on_error (scanners, analysis, normalisation, reporting), plus one collection: line per provider, account or region that could not be collected.

Source code in cloudg/api.py

async def run_pipeline(self, output_dir: str | Path | None = None) -> PipelineResult:
    """Run the complete CloudG pipeline.

    Phases: 1. collection (multi-provider, multi-region), 2. scanning
    (reachability and every enabled scanner), 3. analysis (ontology,
    RAG, Terraform), 4. normalisation, 5. report generation
    (``report.formats``). Files go to ``output_dir``, or
    ``report.output_dir`` when it is None.

    Returns:
        PipelineResult with all outputs. ``errors`` lists every failure
        of the run, each as ``"<phase>: <message>"``: everything
        reported through ``on_error`` (scanners, analysis,
        normalisation, reporting), plus one ``collection:`` line per
        provider, account or region that could not be collected.
    """
    start = time.time()
    out = self._output_dir(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    errors: list[str] = []
    with self._collect_errors(errors):
        return await self._run_pipeline(out, start, errors)

run_pipeline_sync

run_pipeline_sync(output_dir: str | Path | None = None) -> PipelineResult

Synchronous wrapper for run_pipeline(), for scripts and other code that is not already running an event loop.

Raises RuntimeError when an event loop is already running in this thread; use await engine.run_pipeline() there instead.

Source code in cloudg/api.py

def run_pipeline_sync(self, output_dir: str | Path | None = None) -> PipelineResult:
    """Synchronous wrapper for `run_pipeline()`, for scripts and other
    code that is not already running an event loop.

    Raises RuntimeError when an event loop is already running in this
    thread; use ``await engine.run_pipeline()`` there instead.
    """
    return _run_sync(self.run_pipeline(output_dir), "run_pipeline")

collect async

collect() -> CollectionResult

Run multi-provider asset collection.

Returns:

Type	Description
CollectionResult	CollectionResult with assets, edges, and coverage.

Source code in cloudg/api.py

async def collect(self) -> CollectionResult:
    """Run multi-provider asset collection.

    Returns:
        CollectionResult with assets, edges, and coverage.
    """
    self._emit_phase_start("collection")
    start = time.time()

    from cloudg.collectors.multi import MultiAccountCollector

    collector = MultiAccountCollector(self.config)

    try:
        assets, edges, coverage = await collector.collect_all()
    except Exception as exc:
        logger.error("Collection failed: %s", exc)
        self._emit_error("collection", exc)
        return CollectionResult(duration_ms=int((time.time() - start) * 1000))

    result = CollectionResult(
        assets=assets,
        edges=edges,
        coverage=coverage,
        providers_scanned=self.config.providers,
        regions_scanned=getattr(collector, "_resolved_regions", {}),
        duration_ms=int((time.time() - start) * 1000),
    )

    if self.on_collection_complete:
        try:
            self.on_collection_complete(result)
        except Exception:
            logger.debug("Event hook raised; ignoring", exc_info=True)

    return result

scan async

scan(
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    iac_dir: str | None = None,
    images: list[str] | None = None,
    profile: str | None = None,
    output_dir: str | Path | None = None,
) -> list[Finding]

Run security scanners and graph analysis.

Runs all scanners enabled in config.scanners.enabled in parallel, built-in ones and plugins registered under the cloudg.scanners entry point group. Unknown names and scanners whose CLI is not installed are logged and skipped. Without images, Trivy scans the IaC directories instead; the IAM linter runs only when iam is enabled (and there are assets). Each scanner process is killed after config.scanners.timeout_seconds; timeouts and failures are reported through on_error.

Parameters:

Name	Type	Description	Default
assets	list[CloudAsset]	Collected cloud assets.	required
edges	list[NetworkEdge]	Network edges between assets.	required
iac_dir	str | None	IaC directory for Checkov and Trivy (falls back to config).	None
images	list[str] | None	Container images for Trivy (falls back to config).	None
profile	str | None	AWS profile for scanner auth (falls back to the configured AWS auth, aws.profile included).	None
output_dir	str | Path | None	Directory for scanner output files (default: report.output_dir).	None

Returns:

Type	Description
list[Finding]	List of all findings. on_finding receives copies of each scanner's findings as that scanner finishes.

Source code in cloudg/api.py

async def scan(
    self,
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    iac_dir: str | None = None,
    images: list[str] | None = None,
    profile: str | None = None,
    output_dir: str | Path | None = None,
) -> list[Finding]:
    """Run security scanners and graph analysis.

    Runs all scanners enabled in config.scanners.enabled in parallel,
    built-in ones and plugins registered under the ``cloudg.scanners``
    entry point group. Unknown names and scanners whose CLI is not
    installed are logged and skipped. Without images, Trivy scans the
    IaC directories instead; the IAM linter runs only when ``iam`` is
    enabled (and there are assets). Each scanner process is killed
    after config.scanners.timeout_seconds; timeouts and failures are
    reported through ``on_error``.

    Args:
        assets: Collected cloud assets.
        edges: Network edges between assets.
        iac_dir: IaC directory for Checkov and Trivy (falls back to config).
        images: Container images for Trivy (falls back to config).
        profile: AWS profile for scanner auth (falls back to the
            configured AWS auth, aws.profile included).
        output_dir: Directory for scanner output files (default:
            ``report.output_dir``).

    Returns:
        List of all findings. ``on_finding`` receives copies of each
        scanner's findings as that scanner finishes.
    """
    self._emit_phase_start("scanning")
    from cloudg.normaliser import FindingList

    # A FindingList, so Prowler's passing checks reach the normaliser
    all_findings = FindingList()
    out = self._output_dir(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    from cloudg.api_scanners import plan_scanner_jobs

    # Graph-based reachability analysis
    reachability = self._run_reachability_analysis(assets, edges)
    self._emit_findings(reachability)
    all_findings.extend(reachability)

    resolved_iac_dirs = self._resolve_scan_iac_dirs(iac_dir, assets, edges, out)

    # Resolve images
    resolved_images: list[str] = images or list(self.config.scanners.trivy_images)

    plan = plan_scanner_jobs(
        self.config,
        list(self.config.scanners.enabled),
        providers=list(self.config.providers),
        profile=profile,
        out=out,
        assets=assets,
        iac_dirs=resolved_iac_dirs,
        images=resolved_images,
    )
    # Run all scanners concurrently
    scanner_findings = self._run_scan_plan(plan)
    all_findings.extend(scanner_findings)
    all_findings.passed_checks.update(getattr(scanner_findings, "passed_checks", None) or ())

    self._emit_scan_complete(all_findings)

    return all_findings

analyze async

analyze(
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding],
    output_dir: str | Path | None = None,
) -> AnalysisResult

Run ontology, RAG, and Terraform analysis.

Files go to output_dir, or report.output_dir when it is None.

Returns:

Type	Description
AnalysisResult	AnalysisResult with all analysis outputs.

Source code in cloudg/api.py

async def analyze(
    self,
    assets: list[CloudAsset],
    edges: list[NetworkEdge],
    findings: list[Finding],
    output_dir: str | Path | None = None,
) -> AnalysisResult:
    """Run ontology, RAG, and Terraform analysis.

    Files go to ``output_dir``, or ``report.output_dir`` when it is None.

    Returns:
        AnalysisResult with all analysis outputs.
    """
    self._emit_phase_start("analysis")
    out = self._output_dir(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    result = AnalysisResult()

    # Graph
    graph = self._analyze_graph(result, assets, edges)

    # Ontology
    if self.config.ontology.enabled:
        self._analyze_ontology(result, assets, edges, findings, out)

    # RAG export
    if self.config.rag.enabled and graph is not None:
        self._analyze_rag(result, assets, edges, graph, findings, out)

    # Terraform
    if self.config.terraform.enabled:
        self._analyze_terraform(result, assets, edges, out)

    if self.on_analysis_complete:
        try:
            self.on_analysis_complete(result)
        except Exception:
            logger.debug("Event hook raised; ignoring", exc_info=True)

    return result

ingest_reports

ingest_reports(reports: dict[str, list[str | Path]]) -> list[Finding]

Parse pre-existing scanner reports instead of running scanners.

For deployments where Prowler/ScoutSuite/Checkov/Trivy already ran elsewhere (CI, another host, a scheduled job): feed their native output files here and get back cloudg findings, ready for normalise_findings() / analyze().

Parameters:

Name	Type	Description	Default
reports	dict[str, list[str | Path]]	Tool name -> list of report paths (file or directory), e.g. {"prowler": ["./prowler-out/"], "trivy": ["scan.json"]}.	required

Returns:

Type	Description
list[Finding]	Combined list of findings.

Source code in cloudg/api.py

def ingest_reports(self, reports: dict[str, list[str | Path]]) -> list[Finding]:
    """Parse pre-existing scanner reports instead of running scanners.

    For deployments where Prowler/ScoutSuite/Checkov/Trivy already ran
    elsewhere (CI, another host, a scheduled job): feed their native
    output files here and get back cloudg findings, ready for
    `normalise_findings()` / `analyze()`.

    Args:
        reports: Tool name -> list of report paths (file or directory),
            e.g. {"prowler": ["./prowler-out/"], "trivy": ["scan.json"]}.

    Returns:
        Combined list of findings.
    """
    self._emit_phase_start("ingest")
    from cloudg.ingest import ingest_reports as _ingest

    findings = _ingest(reports)
    self._emit_scan_results(findings)
    return findings

normalise_findings

normalise_findings(
    findings: list[Finding], assets: list[CloudAsset] | None = None
) -> ScanResult

Deduplicate, merge across scanners, score, and map to compliance.

Applies the same pipeline run_pipeline() uses: within-scanner dedupe, cross-scanner merging via rules/check_equivalence.yaml, severity scoring, and compliance-framework mapping.

Source code in cloudg/api.py

def normalise_findings(
    self,
    findings: list[Finding],
    assets: list[CloudAsset] | None = None,
) -> ScanResult:
    """Deduplicate, merge across scanners, score, and map to compliance.

    Applies the same pipeline `run_pipeline()` uses: within-scanner
    dedupe, cross-scanner merging via rules/check_equivalence.yaml,
    severity scoring, and compliance-framework mapping.
    """
    from cloudg.normaliser import FindingsNormaliser

    normaliser = FindingsNormaliser(
        rules_dir=self.config.rulesets.rules_dir,
        load_external=self.config.rulesets.load_external,
    )
    return normaliser.normalise(findings, assets=assets or [])

run_from_reports async

run_from_reports(
    reports: dict[str, list[str | Path]], output_dir: str | Path | None = None
) -> PipelineResult

Run the cloudg pipeline on existing scanner outputs.

Needs no cloud access and no scanner binaries. Phases: ingest, normalise, reports (JSON and HTML, as report.formats allows; written to report.output_dir when output_dir is None). Collection does not run, so graph/ontology/Terraform outputs that need live assets are empty; combine with collect() + analyze() when cloud credentials are available.

Returns:

Type	Description
PipelineResult	PipelineResult with findings, scan_result, and report paths.

Source code in cloudg/api.py

async def run_from_reports(
    self,
    reports: dict[str, list[str | Path]],
    output_dir: str | Path | None = None,
) -> PipelineResult:
    """Run the cloudg pipeline on existing scanner outputs.

    Needs no cloud access and no scanner binaries. Phases: ingest,
    normalise, reports (JSON and HTML, as ``report.formats`` allows;
    written to ``report.output_dir`` when ``output_dir`` is None).
    Collection does not run, so
    graph/ontology/Terraform outputs that need live assets are empty;
    combine with `collect()` + `analyze()` when cloud credentials are
    available.

    Returns:
        PipelineResult with findings, scan_result, and report paths.
    """
    start = time.time()
    out = self._output_dir(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    errors: list[str] = []
    with self._collect_errors(errors):
        return self._run_from_reports(reports, out, start, errors)

run_from_reports_sync

run_from_reports_sync(
    reports: dict[str, list[str | Path]], output_dir: str | Path | None = None
) -> PipelineResult

Synchronous wrapper for run_from_reports().

Raises RuntimeError inside a running event loop; await run_from_reports() there instead.

Source code in cloudg/api.py

def run_from_reports_sync(
    self,
    reports: dict[str, list[str | Path]],
    output_dir: str | Path | None = None,
) -> PipelineResult:
    """Synchronous wrapper for `run_from_reports()`.

    Raises RuntimeError inside a running event loop; await
    `run_from_reports()` there instead.
    """
    return _run_sync(self.run_from_reports(reports, output_dir), "run_from_reports")

map_inventory async

map_inventory(
    output_dir: str | Path | None = None,
    findings: list[Finding] | None = None,
    tagging_sweep: bool | None = None,
) -> "Any"

Map the complete infrastructure inventory; no scanners are involved.

Runs the deep inventory collectors (full network fabric plus catch-all sweeps: AWS Resource Groups Tagging API, Azure ARM resources.list, GCP Cloud Asset Inventory) and links every asset into an interconnected map.

Parameters:

Name	Type	Description	Default
output_dir	str | Path | None	When set, exports inventory-map.json / .graphml / inventory-graph.json there.	None
findings	list[Finding] | None	Optional scanner findings produced elsewhere. When given, the overlay is stored on the returned result as asset_map and compliance_map; with output_dir set, asset-map.json and compliance-map.json are exported too.	None
tagging_sweep	bool | None	Override config.inventory.tagging_sweep (False skips the AWS tagging-API sweep).	None

Returns:

Type	Description
'Any'	InventoryResult with assets, edges, coverage, and summary.

Source code in cloudg/api.py

async def map_inventory(
    self,
    output_dir: str | Path | None = None,
    findings: list[Finding] | None = None,
    tagging_sweep: bool | None = None,
) -> "Any":
    """Map the complete infrastructure inventory; no scanners are involved.

    Runs the deep inventory collectors (full network fabric plus
    catch-all sweeps: AWS Resource Groups Tagging API, Azure ARM
    ``resources.list``, GCP Cloud Asset Inventory) and links every
    asset into an interconnected map.

    Args:
        output_dir: When set, exports inventory-map.json / .graphml /
            inventory-graph.json there.
        findings: Optional scanner findings produced elsewhere. When
            given, the overlay is stored on the returned result as
            ``asset_map`` and ``compliance_map``; with output_dir set,
            asset-map.json and compliance-map.json are exported too.
        tagging_sweep: Override config.inventory.tagging_sweep
            (False skips the AWS tagging-API sweep).

    Returns:
        InventoryResult with assets, edges, coverage, and summary.
    """
    self._emit_phase_start("inventory_mapping")
    from cloudg.inventory import InventoryMapper

    mapper = InventoryMapper(self.config, tagging_sweep=tagging_sweep)
    try:
        result = await mapper.map_inventory()
    except Exception as exc:
        logger.error("Inventory mapping failed: %s", exc)
        self._emit_error("inventory_mapping", exc)
        raise

    if findings:
        result.asset_map = mapper.build_asset_map(result, findings)
        result.compliance_map = mapper.build_compliance_map(result, findings)
    if output_dir is not None:
        result.export(output_dir)
        if findings:
            mapper.export_merged(result, findings, output_dir)
    return result

map_inventory_sync

map_inventory_sync(
    output_dir: str | Path | None = None,
    findings: list[Finding] | None = None,
    tagging_sweep: bool | None = None,
) -> "Any"

Synchronous wrapper for map_inventory.

Raises RuntimeError inside a running event loop; await map_inventory there instead.

Source code in cloudg/api.py

def map_inventory_sync(
    self,
    output_dir: str | Path | None = None,
    findings: list[Finding] | None = None,
    tagging_sweep: bool | None = None,
) -> "Any":
    """Synchronous wrapper for :meth:`map_inventory`.

    Raises RuntimeError inside a running event loop; await
    :meth:`map_inventory` there instead.
    """
    return _run_sync(self.map_inventory(output_dir, findings, tagging_sweep), "map_inventory")

Related

• Event hooks

  The six callbacks and when each one fires.

• Result objects

  PipelineResult, CollectionResult and AnalysisResult.

• CloudGConfig

  Building the config the engine runs on.

• The pipeline

  The same phases from the CLI side.

• Python recipes

  Longer integration examples.

• cloudg run

  The command that wraps run_pipeline().