oci-datasafe-adb-inventory/adb_datasafe_inventory.py

#!/usr/bin/env python3
"""Consolidated Autonomous Database x Data Safe inventory."""

from __future__ import annotations

import argparse
import sys

from oci_datasafe_inventory import (
    ProgressBar,
    add_common_args,
    adb_to_base_row,
    build_context,
    enrich_missing_target_associations,
    field,
    get_regions,
    list_all_adbs,
    list_datasafe_targets,
    print_warnings,
    target_associated_adb_id,
    target_to_base_row,
    write_rows,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Gera inventario unico de Autonomous Databases com status de cadastro no Oracle Data Safe."
    )
    add_common_args(parser)
    parser.add_argument(
        "--filter",
        choices=["all", "registered", "not_registered"],
        default="all",
        help="Filtra a saida. Padrao: all.",
    )
    parser.add_argument(
        "--output",
        default="adb_datasafe_inventory.csv",
        help="Arquivo de saida. Padrao: adb_datasafe_inventory.csv",
    )
    parser.add_argument(
        "--split-outputs",
        action="store_true",
        help="Tambem gera adb_in_datasafe.csv e adb_not_in_datasafe.csv a partir do mesmo inventario.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    context = build_context(args)

    if args.all_regions and not context.tenancy_id:
        print("--all-regions exige --tenancy-id ou tenancy configurada no perfil OCI.", file=sys.stderr)
        return 2

    regions = get_regions(
        context.client,
        context.tenancy_id,
        context.default_region,
        args.all_regions,
        args.regions,
    )  # type: ignore[arg-type]
    rows: list[dict[str, object]] = []
    all_errors: list[str] = []

    for region in regions:
        region_label = region or context.default_region or "perfil"
        print(f"Regiao {region_label}: consultando Data Safe...", file=sys.stderr)
        targets, target_errors = list_datasafe_targets(
            context.client,
            context.compartments,
            root_compartment_id=context.tenancy_id,
            region=region,
            workers=args.workers,
            retries=args.retries,
            retry_base_sleep=args.retry_base_sleep,
            include_deleted=args.include_deleted_targets,
        )
        detail_errors = enrich_missing_target_associations(
            context.client,
            targets,
            region=region,
            workers=args.workers,
            retries=args.retries,
            retry_base_sleep=args.retry_base_sleep,
        )
        target_by_adb_id = {
            target_associated_adb_id(target): target
            for target in targets
            if target_associated_adb_id(target)
        }
        print(
            f"Regiao {region_label}: {len(targets)} targets Data Safe encontrados; "
            f"{len(target_by_adb_id)} com OCID de ADB associado; "
            f"varrendo {len(context.compartments)} compartments...",
            file=sys.stderr,
        )

        progress = ProgressBar(
            f"Regiao {region_label}",
            len(context.compartments),
            enabled=not args.no_progress,
        )
        stats = {"adbs": 0, "reg": 0, "nao_reg": 0, "erros": 0}

        def update_progress(compartment: dict[str, object], items: list[dict[str, object]], error: str | None) -> None:
            stats["adbs"] += len(items)
            stats["reg"] += sum(1 for adb in items if field(adb, "id") in target_by_adb_id)
            stats["nao_reg"] += sum(1 for adb in items if field(adb, "id") not in target_by_adb_id)
            if error:
                stats["erros"] += 1
            progress.update(
                current_label=str(field(compartment, "name", "display-name", default=field(compartment, "id"))),
                stats=stats,
            )

        adbs, adb_errors = list_all_adbs(
            context.client,
            context.compartments,
            region=region,
            workers=args.workers,
            retries=args.retries,
            retry_base_sleep=args.retry_base_sleep,
            include_terminated=args.include_terminated_adbs,
            progress_callback=update_progress,
        )
        progress.finish(stats=stats)
        all_errors.extend(target_errors + detail_errors + adb_errors)

        for adb in sorted(adbs, key=lambda item: (field(item, "display-name"), field(item, "id"))):
            adb_id = field(adb, "id")
            target = target_by_adb_id.get(adb_id)
            registered = target is not None
            if args.filter == "registered" and not registered:
                continue
            if args.filter == "not_registered" and registered:
                continue

            row = adb_to_base_row(adb, context.compartment_index, region=region)
            row["datasafe_registered"] = str(registered).lower()
            if target:
                row.update(target_to_base_row(target, context.compartment_index))
            rows.append(row)

    write_rows(rows, args.output, args.format)
    registered_rows = [row for row in rows if row.get("datasafe_registered") == "true"]
    not_registered_rows = [row for row in rows if row.get("datasafe_registered") == "false"]

    if args.split_outputs:
        write_rows(registered_rows, "adb_in_datasafe.csv", args.format)
        write_rows(not_registered_rows, "adb_not_in_datasafe.csv", args.format)

    print(f"ADBs no inventario: {len(rows)}")
    print(f"ADBs cadastrados no Data Safe: {len(registered_rows)}")
    print(f"ADBs nao cadastrados no Data Safe: {len(not_registered_rows)}")
    print(f"Arquivo gerado: {args.output}")
    if args.split_outputs:
        print("Arquivos adicionais gerados: adb_in_datasafe.csv, adb_not_in_datasafe.csv")
    print_warnings("o levantamento", all_errors)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())