refactor: scope CSV updates through asset facades

Author: ayush-shah

dynamic_csv_importer/README.md

@@ -4,9 +4,10 @@
 > This example targets `openmetadata-ingestion==2.0.0.0rc1`. It updates existing assets and does
 > not create entities or columns.
 
-The importer matches the `column.name*` field from an OpenMetadata CSV export against columns on
-tables and dashboard data models. It can update column descriptions and display names or run in a
-report-only mode.
+The importer matches the `column.name*` field from an OpenMetadata CSV export
+against columns on explicitly named tables and dashboard data models. It can
+update column descriptions and display names or build the same plan in
+report-only mode. It never scans the catalog.
 
 ## Run the Example
 
@@ -19,14 +20,17 @@ export OPENMETADATA_HOST="http://localhost:8585/api"
 export OPENMETADATA_JWT_TOKEN="<personal-access-token>"
 python -m dynamic_csv_importer.csv_importer \
   --csv-path dynamic_csv_importer/sample_dbt_jaffle.csv \
-  --entities all \
+  --table-fqn sample.database.schema.customers \
+  --dashboard-data-model-fqn sample_dashboard.customer_model \
   --report-only
 ```
 
-Remove `--report-only` after reviewing the match counts. Override the environment-based connection
-settings with `--url` and `--jwt-token` when needed.
+Both FQN flags are repeatable, and at least one is required. The explicit FQNs
+are the authorization boundary: only those named assets can be updated. Remove
+`--report-only` after reviewing the counts; live mutation is the default.
 
-Supported `--entities` values are `all`, `Table`, and `DashboardDataModel`.
+Credentials are read only from `OPENMETADATA_HOST` and
+`OPENMETADATA_JWT_TOKEN`. There are no URL or token CLI flags.
 
 ## CSV Columns
 
@@ -39,5 +43,17 @@ remaining fields make the input compatible with a normal OpenMetadata export.
 
 ## Validation Boundary
 
-Repository tests validate argument handling, CSV parsing, matching, and request-model compatibility
-without connecting to OpenMetadata.
+The complete CSV and every target are loaded before the first update. Invalid
+CSV rows or a missing target abort the run before mutation. Each destination is
+a deep copy and is sent through `Tables.update()` or
+`DashboardDataModels.update()`.
+
+There is a normal concurrency window between the initial reads and facade
+updates. Each facade update re-reads the current entity before creating its JSON
+Patch, but the planned destination came from the earlier snapshot. Use
+`--report-only`, rerun against disposable targets first, and avoid concurrent
+edits to the same columns during a live import.
+
+Repository tests validate arguments, full CSV parsing, deep-copy changes,
+preflight ordering, and mocked facade updates without connecting to
+OpenMetadata.

dynamic_csv_importer/csv_importer.py

@@ -1,41 +1,23 @@
-"""Update existing OpenMetadata column descriptions and display names from CSV."""
+"""Update named OpenMetadata assets from a fully parsed column CSV."""
+
+from __future__ import annotations
 
 import argparse
 import csv
 import logging
 import os
-from collections.abc import Generator, Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
-from enum import Enum
 from pathlib import Path
+from typing import Any
 
-from metadata.generated.schema.entity.data.dashboardDataModel import (
-    DashboardDataModel,
-)
-from metadata.generated.schema.entity.data.table import Table
-from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (  # noqa: E501
-    AuthProvider,
-    OpenMetadataConnection,
-)
-from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
-    OpenMetadataJWTClientConfig,
-)
 from metadata.generated.schema.type.basic import Markdown
-from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.ometa.utils import model_str
+from metadata.sdk import DashboardDataModels, Tables, configure, reset
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
 logger = logging.getLogger(__name__)
-
-ALL_ENTITIES = [Table, DashboardDataModel]
-
-
-class SupportedEntities(Enum):
-    """Entity collections supported by the importer."""
-
-    ALL = "all"
-    TABLES = Table.__name__
-    DASHBOARD_DATA_MODEL = DashboardDataModel.__name__
+REQUIRED_ENV = ("OPENMETADATA_HOST", "OPENMETADATA_JWT_TOKEN")
 
 
 class CSVColumnSchema(BaseModel):
@@ -65,39 +47,42 @@ class ProcessSummary:
 
 
 def parse_arguments(argv: Sequence[str] | None = None) -> argparse.Namespace:
-    """Parse CLI options, using standard OpenMetadata environment variables."""
-    host = os.environ.get("OPENMETADATA_HOST")
-    token = os.environ.get("OPENMETADATA_JWT_TOKEN")
+    """Parse explicit asset targets without accepting credentials on the CLI."""
     parser = argparse.ArgumentParser(
-        description="Import column metadata from CSV into OpenMetadata"
+        description="Import column metadata into explicitly named OpenMetadata assets"
     )
     parser.add_argument(
         "--csv-path", required=True, help="Path to the CSV file containing metadata"
     )
     parser.add_argument(
-        "--entities",
-        choices=[entity.value for entity in SupportedEntities],
-        default=SupportedEntities.ALL.value,
-        help="Entity types to process (default: all)",
-    )
-    parser.add_argument(
-        "--url",
-        default=host,
-        required=host is None,
-        help="OpenMetadata API URL; defaults to OPENMETADATA_HOST",
+        "--table-fqn",
+        action="append",
+        default=[],
+        help="Table FQN to update; repeat for multiple tables",
     )
     parser.add_argument(
-        "--jwt-token",
-        default=token,
-        required=token is None,
-        help="JWT token; defaults to OPENMETADATA_JWT_TOKEN",
+        "--dashboard-data-model-fqn",
+        action="append",
+        default=[],
+        help="Dashboard data model FQN to update; repeat for multiple models",
     )
     parser.add_argument(
         "--report-only",
         action="store_true",
-        help="Report matches without updating entities",
+        help="Build and report the update plan without mutating assets",
     )
-    return parser.parse_args(argv)
+    args = parser.parse_args(argv)
+    if not args.table_fqn and not args.dashboard_data_model_fqn:
+        parser.error("at least one --table-fqn or --dashboard-data-model-fqn is required")
+    return args
+
+
+def validate_environment(environ: Mapping[str, str] | None = None) -> None:
+    """Require standard credentials before configuring the SDK."""
+    values = os.environ if environ is None else environ
+    missing = [name for name in REQUIRED_ENV if not values.get(name)]
+    if missing:
+        raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
 
 
 def clean_column_name(column_name: str) -> str:
@@ -108,54 +93,25 @@ def clean_column_name(column_name: str) -> str:
 
 
 def load_csv_data(csv_path: str | Path) -> dict[str, CSVColumnSchema]:
-    """Load valid rows from an OpenMetadata column CSV export."""
+    """Parse and validate the complete CSV before any server request."""
     rows: dict[str, CSVColumnSchema] = {}
+    failures: list[int] = []
     with Path(csv_path).open(encoding="utf-8", newline="") as file:
         for row_number, raw_row in enumerate(csv.DictReader(file), 2):
             try:
                 row = CSVColumnSchema.model_validate(raw_row)
-            except ValidationError as exc:
-                logger.warning("Skipping invalid CSV row %d: %s", row_number, exc)
+            except ValidationError:
+                failures.append(row_number)
                 continue
             row.column_name = clean_column_name(row.column_name)
             rows[row.column_name] = row
+    if failures:
+        lines = ", ".join(str(number) for number in failures)
+        raise ValueError(f"Invalid CSV rows: {lines}")
     return rows
 
 
-def initialize_ometa_client(url: str, jwt_token: str) -> OpenMetadata:
-    """Construct the OpenMetadata 2.0 client without performing API calls."""
-    config = OpenMetadataConnection(
-        hostPort=url,
-        authProvider=AuthProvider.openmetadata,
-        securityConfig=OpenMetadataJWTClientConfig(jwtToken=jwt_token),
-    )
-    return OpenMetadata(config=config)
-
-
-def _iter_entity_type(ometa: OpenMetadata, entity_type: type) -> Generator:
-    """Yield one entity type through the paginated API."""
-    after = None
-    while True:
-        page = ometa.list_entities(entity=entity_type, limit=100, after=after)
-        yield from page.entities
-        if not page.after:
-            return
-        after = page.after
-
-
-def get_entities_to_process(ometa: OpenMetadata, entities: str) -> Generator:
-    """Yield the selected supported entities."""
-    if entities == SupportedEntities.ALL.value:
-        entity_types = ALL_ENTITIES
-    elif entities == SupportedEntities.TABLES.value:
-        entity_types = [Table]
-    else:
-        entity_types = [DashboardDataModel]
-    for entity_type in entity_types:
-        yield from _iter_entity_type(ometa, entity_type)
-
-
-def matching_rows(entity, csv_data: dict[str, CSVColumnSchema]) -> list[CSVColumnSchema]:
+def matching_rows(entity: Any, csv_data: dict[str, CSVColumnSchema]) -> list[CSVColumnSchema]:
     """Return CSV rows that match columns on an entity."""
     if not getattr(entity, "columns", None):
         return []
@@ -166,8 +122,8 @@ def matching_rows(entity, csv_data: dict[str, CSVColumnSchema]) -> list[CSVColum
     ]
 
 
-def build_updated_entity(entity, matches: list[CSVColumnSchema]):
-    """Return an updated entity copy, or None when no value changes."""
+def build_updated_entity(entity: Any, matches: list[CSVColumnSchema]) -> Any | None:
+    """Return a deep-copied destination, or None when no value changes."""
     updated_entity = entity.model_copy(deep=True)
     updates_made = False
     rows_by_name = {row.column_name: row for row in matches}
@@ -190,44 +146,70 @@ def build_updated_entity(entity, matches: list[CSVColumnSchema]):
     return updated_entity if updates_made else None
 
 
+def _unique(values: Sequence[str]) -> list[str]:
+    return list(dict.fromkeys(values))
+
+
 def process_entities(
-    ometa: OpenMetadata,
-    entities: str,
+    table_fqns: Sequence[str],
+    dashboard_data_model_fqns: Sequence[str],
     csv_data: dict[str, CSVColumnSchema],
     report_only: bool = False,
 ) -> ProcessSummary:
-    """Match and optionally patch entities, returning deterministic counts."""
-    processed = matched = total_matches = updated = 0
-    for entity in get_entities_to_process(ometa, entities):
-        processed += 1
+    """Retrieve every authorized target, build all changes, then update."""
+    targets = [(Tables, fqn) for fqn in _unique(table_fqns)]
+    targets += [(DashboardDataModels, fqn) for fqn in _unique(dashboard_data_model_fqns)]
+    retrieved: list[tuple[Any, Any]] = []
+    for facade, fqn in targets:
+        entity = facade.retrieve_by_name(fqn, nullable=True)
+        if entity is None:
+            raise RuntimeError(f"Target not found: {fqn}")
+        retrieved.append((facade, entity))
+
+    matched = total_matches = 0
+    planned: list[tuple[Any, Any]] = []
+    for facade, entity in retrieved:
         matches = matching_rows(entity, csv_data)
         if not matches:
             continue
         matched += 1
         total_matches += len(matches)
         destination = build_updated_entity(entity, matches)
-        if not report_only and destination is not None:
-            result = ometa.patch(
-                entity=type(entity),
-                source=entity,
-                destination=destination,
-                skip_on_failure=False,
-            )
-            if result is not None:
+        if destination is not None:
+            planned.append((facade, destination))
+
+    updated = 0
+    if not report_only:
+        for facade, destination in planned:
+            if facade.update(destination) is not None:
                 updated += 1
-    return ProcessSummary(processed, matched, total_matches, updated)
+    return ProcessSummary(len(retrieved), matched, total_matches, updated)
 
 
 def main(argv: Sequence[str] | None = None) -> int:
-    """Run the importer CLI."""
-    logging.basicConfig(
-        level=logging.INFO,
-        format="[%(asctime)s] %(levelname)-8s - %(message)s",
-    )
-    args = parse_arguments(argv)
-    csv_data = load_csv_data(args.csv_path)
-    client = initialize_ometa_client(args.url, args.jwt_token)
-    summary = process_entities(client, args.entities, csv_data, args.report_only)
+    """Run the importer with environment-only credentials."""
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    try:
+        args = parse_arguments(argv)
+        validate_environment()
+        csv_data = load_csv_data(args.csv_path)
+        configure()
+        try:
+            summary = process_entities(
+                args.table_fqn,
+                args.dashboard_data_model_fqn,
+                csv_data,
+                args.report_only,
+            )
+        finally:
+            reset()
+    except (ValueError, RuntimeError) as exc:
+        logger.error("CSV import failed: %s", exc)
+        return 1
+    except Exception as exc:
+        logger.error("CSV import failed: %s", type(exc).__name__)
+        return 1
+
     logger.info("Entities processed: %d", summary.entities_processed)
     logger.info("Entities with matches: %d", summary.entities_with_matches)
     logger.info("Column matches: %d", summary.total_matches)

dynamic_csv_importer/sample_dbt_jaffle.csv

@@ -13,4 +13,4 @@ credit_card_amount,Credit Card New Amount,Amount paid using credit card,DECIMAL,
 coupon_amount,Coupon Amount,Discount amount applied through coupon codes,DECIMAL,DOUBLE,,,Finance.Payment,Payment.Financial
 bank_transfer_amount,Bank Transfer Amount,Amount paid via bank transfer new,DECIMAL,DOUBLE,,,Finance.Payment,Payment.Financial
 gift_card_amount,Gift Card Amount,Amount paid using gift card new balance,DECIMAL,DOUBLE,,,Finance.Payment,Payment.Financial
-country_name,Country Super Name,this is a cool new description,,,,,,
+country_name,Country Super Name,this is a cool new description,VARCHAR,STRING,,,,