Initial OCI Data Safe ADB inventory tool

adb_not_in_datasafe.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Report Autonomous Databases that are not 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,
+    write_rows,
+)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Lista Autonomous Databases que nao estao cadastrados no Oracle Data Safe."
+    )
+    add_common_args(parser)
+    parser.add_argument(
+        "--output",
+        default="adb_not_in_datasafe.csv",
+        help="Arquivo de saida. Padrao: adb_not_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)
+
+        for adb in sorted(adbs, key=lambda item: (field(item, "display-name"), field(item, "id"))):
+            if field(adb, "id") in registered_adb_ids:
+                continue
+            row = adb_to_base_row(adb, context.compartment_index, region=region)
+            row["datasafe_registered"] = "false"
+            rows.append(row)
+
+    write_rows(rows, args.output, args.format)
+    print(f"ADBs nao 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())