File size: 1,781 Bytes
6bff5d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | """CatalogValidator — Pydantic + business-rule validation for a catalog.
Pydantic handles shape; this layer adds invariants that span fields.
"""
from .models import Catalog
class CatalogValidationError(Exception):
pass
class CatalogValidator:
"""Validates a Catalog beyond Pydantic schema checks.
Business rules:
- All source_ids unique within a user
- All table_ids unique within a source
- All column_ids unique within a table
- foreign_keys (when added) reference existing tables/columns
"""
def validate(self, catalog: Catalog) -> None:
seen_sources: set[str] = set()
for source in catalog.sources:
if source.source_id in seen_sources:
raise CatalogValidationError(
f"duplicate source_id {source.source_id!r} in catalog "
f"for user_id={catalog.user_id!r}"
)
seen_sources.add(source.source_id)
seen_tables: set[str] = set()
for table in source.tables:
if table.table_id in seen_tables:
raise CatalogValidationError(
f"duplicate table_id {table.table_id!r} in source "
f"{source.source_id!r}"
)
seen_tables.add(table.table_id)
seen_columns: set[str] = set()
for column in table.columns:
if column.column_id in seen_columns:
raise CatalogValidationError(
f"duplicate column_id {column.column_id!r} in table "
f"{table.table_id!r} (source {source.source_id!r})"
)
seen_columns.add(column.column_id)
|