#!/usr/bin/env python3 """Fetch Boomi Data Integration data flow details and save them to JSON files.""" from __future__ import annotations import argparse import datetime as dt import json import os import re import sys import urllib.parse import urllib.request from typing import Any, Iterable DEFAULT_BASE_URL = "https://api.rivery.io/v1" def build_url(base_url: str, account_id: str, environment_id: str, river_cross_id: str) -> str: safe_account_id = urllib.parse.quote(account_id, safe="") safe_environment_id = urllib.parse.quote(environment_id, safe="") safe_river_cross_id = urllib.parse.quote(river_cross_id, safe="") return ( f"{base_url}/accounts/{safe_account_id}/environments/{safe_environment_id}" f"/rivers/{safe_river_cross_id}" ) def fetch_page(url: str, token: str) -> dict[str, Any]: request = urllib.request.Request( url, headers={ "Authorization": f"Bearer {token}", "Accept": "application/json", }, ) try: with urllib.request.urlopen( # nosec B310 - controlled endpoint request ) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: details = exc.read().decode("utf-8").strip() message = f"API request failed with HTTP {exc.code}." if details: message = f"{message} Response: {details}" raise RuntimeError(message) from exc def load_river_ids(payload: dict[str, Any]) -> list[str]: data = payload.get("data", payload) items = data.get("items") if isinstance(data, dict) else None if items is None: raise ValueError("Expected 'data.items' in the rivers JSON payload.") if not isinstance(items, list): raise ValueError("Expected 'data.items' to be a list in the rivers JSON payload.") river_ids = [] for item in items: if not isinstance(item, dict): continue if item.get("is_api_v2") is not True: continue river_cross_id = item.get("river_cross_id") if river_cross_id: river_ids.append(str(river_cross_id)) return river_ids def write_output(path: str, payload: dict[str, Any]) -> None: output_dir = os.path.dirname(path) if output_dir: os.makedirs(output_dir, exist_ok=True) with open(path, "w", encoding="utf-8") as handle: json.dump(payload, handle, indent=2, ensure_ascii=False) handle.write("\n") def sanitize_filename(value: str) -> str: cleaned = value.strip() cleaned = re.sub(r"[<>:\"/\\\\|?*\x00-\x1f]", "-", cleaned) cleaned = cleaned.replace(os.path.sep, "-") cleaned = re.sub(r"\s+", " ", cleaned) cleaned = cleaned.strip(" .") return cleaned or "unknown" def build_output_filename(template: str, river_cross_id: str, river_name: str | None) -> str: safe_id = sanitize_filename(river_cross_id) safe_name = sanitize_filename(river_name or "unknown") try: filename = template.format(river_cross_id=safe_id, river_name=safe_name) except KeyError as exc: raise ValueError( "Output filename template must use {river_cross_id} and/or {river_name}." ) from exc return sanitize_filename(filename) def payload_matches_existing(path: str, payload: dict[str, Any]) -> bool: if not os.path.exists(path): return False try: existing = load_payload(path) except (OSError, json.JSONDecodeError): return False if not isinstance(existing, dict): return False existing_copy = dict(existing) existing_copy.pop("generated_at", None) payload_copy = dict(payload) payload_copy.pop("generated_at", None) return existing_copy == payload_copy def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Fetch Boomi Data Integration data flow details for each river_cross_id and " "save them to JSON files." ) ) parser.add_argument( "--input", default=os.path.join("dataflows", "api_v2_dataflows.json"), help=( "Path to the data flows list JSON file (default: " "dataflows/api_v2_dataflows.json)." ), ) parser.add_argument( "--output-dir", default="dataflow-details", help=( "Directory to store per-data-flow config folders (default: dataflow-details)." ), ) parser.add_argument( "--output-filename", default="{river_cross_id} {river_name}.json", help=( "Filename template for river configs (default: " "'{river_cross_id} {river_name}.json')." ), ) parser.add_argument( "--account-id", default=( os.getenv("BOOMI_ACCOUNT_ID") or os.getenv("BOOMI_ACCOUNT_ID_SECRET") or os.getenv("RIVERY_ACCOUNT_ID") or os.getenv("RIVERY_ACCOUNT_ID_SECRET") ), help=( "Boomi account ID (or set BOOMI_ACCOUNT_ID / BOOMI_ACCOUNT_ID_SECRET, " "or legacy RIVERY_ACCOUNT_ID / RIVERY_ACCOUNT_ID_SECRET)." ), ) parser.add_argument( "--environment-id", default=( os.getenv("BOOMI_ENVIRONMENT_ID") or os.getenv("BOOMI_ENVIRONMENT_ID_SECRET") or os.getenv("RIVERY_ENVIRONMENT_ID") or os.getenv("RIVERY_ENVIRONMENT_ID_SECRET") ), help=( "Boomi environment ID (or set BOOMI_ENVIRONMENT_ID / " "BOOMI_ENVIRONMENT_ID_SECRET, or legacy RIVERY_ENVIRONMENT_ID / " "RIVERY_ENVIRONMENT_ID_SECRET)." ), ) parser.add_argument( "--token", default=( os.getenv("BOOMI_API_TOKEN") or os.getenv("BOOMI_API_TOKEN_SECRET") or os.getenv("RIVERY_API_TOKEN") or os.getenv("RIVERY_API_TOKEN_SECRET") ), help=( "Boomi API token (or set BOOMI_API_TOKEN / BOOMI_API_TOKEN_SECRET, " "or legacy RIVERY_API_TOKEN / RIVERY_API_TOKEN_SECRET)." ), ) parser.add_argument( "--base-url", default=DEFAULT_BASE_URL, help=f"API base URL (default: {DEFAULT_BASE_URL}).", ) return parser.parse_args(argv) def validate_args(args: argparse.Namespace) -> None: missing = [] if not args.account_id: missing.append("--account-id or BOOMI_ACCOUNT_ID (or legacy RIVERY_ACCOUNT_ID)") if not args.environment_id: missing.append("--environment-id or BOOMI_ENVIRONMENT_ID (or legacy RIVERY_ENVIRONMENT_ID)") if not args.token: missing.append("--token or BOOMI_API_TOKEN (or legacy RIVERY_API_TOKEN)") if missing: raise ValueError(f"Missing required settings: {', '.join(missing)}") def load_payload(path: str) -> dict[str, Any]: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def resolve_metadata(args: argparse.Namespace, payload: dict[str, Any]) -> tuple[str, str]: account_id = args.account_id or payload.get("account_id") environment_id = args.environment_id or payload.get("environment_id") args.account_id = account_id args.environment_id = environment_id validate_args(args) return account_id, environment_id def fetch_river_details( river_ids: Iterable[str], token: str, base_url: str, account_id: str, environment_id: str, ) -> list[tuple[str, dict[str, Any]]]: results = [] for river_cross_id in river_ids: url = build_url(base_url, account_id, environment_id, river_cross_id) payload = fetch_page(url, token) results.append((river_cross_id, payload)) return results def main(argv: list[str]) -> int: args = parse_args(argv) payload = load_payload(args.input) account_id, environment_id = resolve_metadata(args, payload) river_ids = load_river_ids(payload) os.makedirs(args.output_dir, exist_ok=True) results = fetch_river_details(river_ids, args.token, args.base_url, account_id, environment_id) updated = 0 skipped = 0 for river_cross_id, data in results: river_name = None if isinstance(data, dict): river_name = data.get("name") output_payload = { "source": "Boomi Data Integration API", "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(), "account_id": account_id, "environment_id": environment_id, "river_cross_id": river_cross_id, "data": data, } filename = build_output_filename(args.output_filename, river_cross_id, river_name) output_path = os.path.join(args.output_dir, filename) if payload_matches_existing(output_path, output_payload): skipped += 1 continue write_output(output_path, output_payload) updated += 1 print( f"Saved {updated} data flow detail files to {args.output_dir} " f"({skipped} unchanged)." ) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))