oci-datasafe-adb-inventory/adb_in_datasafe.py

#!/usr/bin/env python3
"""Report Autonomous Databases that are registered as Data Safe targets."""

from __future__ import annotations

import argparse
import sys

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


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Lista Autonomous Databases cadastrados no Oracle Data Safe e seus compartments."
    )
    add_common_args(parser)
    parser.add_argument(
        "--output",
        default="adb_in_datasafe.csv",
        help="Arquivo de saida. Padrao: adb_in_datasafe.csv",
    )
    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,
        )
        registered_adb_ids = {
            target_associated_adb_id(target)
            for target in targets
            if target_associated_adb_id(target)
        }
        print(
            f"Regiao {region_label}: {len(targets)} targets Data Safe encontrados; "
            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 registered_adb_ids)
            stats["nao_reg"] += sum(1 for adb in items if field(adb, "id") not in registered_adb_ids)
            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)

        adb_by_id = {field(adb, "id"): adb for adb in adbs if field(adb, "id")}

        for target in sorted(targets, key=lambda item: (field(item, "display-name"), field(item, "id"))):
            adb_id = target_associated_adb_id(target)
            adb = adb_by_id.get(adb_id)
            if adb:
                row = adb_to_base_row(adb, context.compartment_index, region=region)
                row["adb_found_in_inventory"] = "true"
            else:
                row = {
                    "region": region or "",
                    "adb_display_name": "",
                    "adb_db_name": "",
                    "adb_ocid": adb_id,
                    "adb_lifecycle_state": "",
                    "adb_workload": "",
                    "adb_is_free_tier": "",
                    "adb_time_created": "",
                    "adb_compartment_name": "",
                    "adb_compartment_path": "",
                    "adb_compartment_ocid": "",
                    "adb_found_in_inventory": "false",
                }
            row.update(target_to_base_row(target, context.compartment_index))
            row["datasafe_registered"] = "true"
            rows.append(row)

    write_rows(rows, args.output, args.format)
    print(f"ADBs cadastrados no Data Safe: {len(rows)}")
    print(f"Arquivo gerado: {args.output}")
    print_warnings("o levantamento", all_errors)
    return 0


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