diff --git a/submission/sample-candidate/approach_explanation.txt b/submission/sample-candidate/approach_explanation.txt new file mode 100644 index 0000000..f3eb04d --- /dev/null +++ b/submission/sample-candidate/approach_explanation.txt @@ -0,0 +1,42 @@ +Pull Request: Technical Approach for CDC Pipeline +What is this? +Before writing the final code for the assignment, this dummy PR maps out the technical plan for building a reliable Change Data Capture (CDC) platform using Python and DuckDB, based on the criteria in the DATA_ASSIGNMENT.md. + + +How it Works (The 4 Layers) +1. The Source System (source_db.py) +Instead of a single flat file, we use a proper relational setup (like customers and orders). + +Strict Rules: Financial data uses exact DECIMAL(18,2) types instead of unsafe floats, and statuses are restricted using database enums. + +2. The Ingestion Guard (cdc_pipeline.py) +This moves data safely from the source into the lakehouse. + +Fail-Closed Schema Check: Before pulling data, the pipeline verifies the source structure against a contract. If a column is dropped or changed upstream, the pipeline stops immediately (SchemaMismatchException) so it doesn't corrupt anything downstream. + +Checkpoints: It tracks data using timestamps. If the pipeline crashes mid-run, it resumes exactly where it left off without duplicating data or creating gaps. + +3. The Data Lake: Absolute History (lake_append.py) +The Data Lake is an append-only ledger. It never updates or overwrites records. Every single change (Insert, Update, Delete) creates a new row stamped with metadata: + +_cdc_op_type: Was it an (I)nsert, (U)pdate, or (D)elete? + +_cdc_extracted_at: When did we pull it? + +_cdc_lsn_version: The exact sequence order. + +4. The Warehouse: Clean Analytics & Time Travel (warehouse_snapshot.py) +The Warehouse provides a clean view of the latest data for end users. + +The Latest State: It reads the lake's history and runs a window function (ROW_NUMBER() OVER) to grab only the latest update for each record. Deleted rows are filtered out automatically. + +Time Travel: Because the lake contains an unbroken history of changes, you can view exactly what the warehouse looked like at any point in the past just by filtering the query by a target timestamp (WHERE _cdc_extracted_at <= target_time). + +Testing the Edge Cases +The test suite skips basic checks and focuses heavily on failure recovery: + +The Lifecycle Test: Inserts, updates, and deletes a record. It ensures the lake records all 3 events, while the warehouse correctly shows 0 active rows. + +The Circuit Breaker Test: Drops a source column to prove the pipeline actively stops running. + +The Time-Travel Test: Queries the warehouse at a specific historical timestamp to verify it rolls back accurately. \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/cdc.py b/submission/sample-candidate/pipeline/cdc.py index c8c2dcb..b799fd5 100644 --- a/submission/sample-candidate/pipeline/cdc.py +++ b/submission/sample-candidate/pipeline/cdc.py @@ -1,89 +1,48 @@ -""" -CDC capture layer. - -Simulates WAL-based change capture: every insert/update/delete on the source -produces a CDCRecord with a monotonically increasing sequence number. - -Replay safety: callers can checkpoint the last processed sequence and call -records_since(offset) to replay only unprocessed changes after a restart. -""" - -from __future__ import annotations - -from dataclasses import dataclass, field -from datetime import datetime, timezone -from typing import Any - -VALID_OPERATIONS = frozenset({"insert", "update", "delete"}) - - -@dataclass -class CDCRecord: - operation: str - table: str - primary_key: str - data: dict[str, Any] - captured_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) - sequence: int = 0 - - def __post_init__(self) -> None: - if self.operation not in VALID_OPERATIONS: - raise ValueError( - f"Invalid CDC operation {self.operation!r}. " - f"Must be one of: {sorted(VALID_OPERATIONS)}" - ) - - -class CDCCapture: - """ - In-process CDC log. - - Production analogue: a Debezium/Kafka connector reading Postgres WAL. - Each record carries a sequence number equivalent to a Kafka offset or - Postgres LSN for checkpoint-based replay. - """ - - def __init__(self) -> None: - self._log: list[CDCRecord] = [] - self._seq: int = 0 - - # ── public write API ───────────────────────────────────────────────────── - - def insert(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("insert", table, pk, data) - - def update(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("update", table, pk, data) - - def delete(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord: - return self._record("delete", table, pk, data) - - # ── public read / replay API ───────────────────────────────────────────── - - def records_since(self, offset: int = 0) -> list[CDCRecord]: - """Return all records with sequence > offset (checkpoint replay).""" - return [r for r in self._log if r.sequence > offset] - - @property - def latest_sequence(self) -> int: - return self._seq - - @property - def log(self) -> list[CDCRecord]: - return list(self._log) - - # ── internal ───────────────────────────────────────────────────────────── - - def _record( - self, operation: str, table: str, pk: str, data: dict[str, Any] - ) -> CDCRecord: - self._seq += 1 - rec = CDCRecord( - operation=operation, - table=table, - primary_key=pk, - data=data, - sequence=self._seq, - ) - self._log.append(rec) - return rec +import datetime +from source.models import SCHEMA_CONTRACT + +class SchemaMismatchException(Exception): pass + +class CDCPipeline: + def __init__(self, source_conn, lake_conn): + self.source_conn = source_conn + self.lake_conn = lake_conn + self.high_watermarks = {table: datetime.datetime.min for table in SCHEMA_CONTRACT.keys()} + + def verify_schema_contract(self): + """Active safety sentinel. Throws exception if upstream columns are corrupted.""" + for table_name, expected_cols in SCHEMA_CONTRACT.items(): + actual_cols = self.source_conn.execute(f"PRAGMA table_info('{table_name}')").fetchall() + actual_col_names = [col[1] for col in actual_cols] + + for col in expected_cols: + if col not in actual_col_names: + raise SchemaMismatchException( + f"CRITICAL DRIFT DETECTED: Column '{col}' is missing on source table '{table_name}'!" + ) + + def ingest_table_changes(self, table_name: str, lsn: int): + """Pulls incremental change logs past the last recorded high-watermark.""" + self.verify_schema_contract() + now = datetime.datetime.now() + + time_col = "created_at" if table_name == "transactions" else "updated_at" + cols_str = ", ".join(SCHEMA_CONTRACT[table_name]) + + changes = self.source_conn.execute(f""" + SELECT {cols_str} FROM {table_name} WHERE {time_col} > ? ORDER BY {time_col} ASC + """, (self.high_watermarks[table_name],)).fetchall() + + if not changes: + return + + for row in changes: + # Map mutations safely. If it's the initial pull, flag as Insert; otherwise Update + op_type = 'I' if self.high_watermarks[table_name] == datetime.datetime.min else 'U' + placeholders = ", ".join(["?"] * (len(row) + 3)) + + self.lake_conn.execute(f""" + INSERT INTO lake_{table_name}_history VALUES ({placeholders}) + """, (*row, op_type, now, lsn)) + + self.high_watermarks[table_name] = max([row[-1] for row in changes]) \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/lake.py b/submission/sample-candidate/pipeline/lake.py index 7cb15a7..9292aa4 100644 --- a/submission/sample-candidate/pipeline/lake.py +++ b/submission/sample-candidate/pipeline/lake.py @@ -1,69 +1,22 @@ -""" -Lake layer — append-only storage for every CDC event. - -Every change is written exactly once. The lake is the source of truth for -point-in-time replay and historical reconstruction. - -Production analogue: Parquet/Delta files on S3 or GCS, partitioned by -table_name and captured_at date. No row is ever modified or deleted. -""" - -from __future__ import annotations - -import json -from datetime import datetime - import duckdb - -from pipeline.cdc import CDCRecord - - -def create_lake_table(conn: duckdb.DuckDBPyConnection) -> None: - conn.execute(""" - CREATE TABLE IF NOT EXISTS lake_cdc_events ( - sequence INTEGER NOT NULL, - operation VARCHAR NOT NULL, - table_name VARCHAR NOT NULL, - primary_key VARCHAR NOT NULL, - data VARCHAR NOT NULL, - captured_at TIMESTAMP NOT NULL - ) - """) - - -def append_to_lake(conn: duckdb.DuckDBPyConnection, records: list[CDCRecord]) -> int: - """ - Append CDC records to the lake. - - Returns the number of records written. - Idempotency note: in production, deduplicate by sequence before appending. - """ - if not records: - return 0 - - rows = [ - ( - r.sequence, - r.operation, - r.table, - r.primary_key, - _serialize(r.data), - r.captured_at, - ) - for r in records - ] - conn.executemany( - "INSERT INTO lake_cdc_events VALUES (?, ?, ?, ?, ?, ?)", - rows, - ) - return len(rows) - - -def _serialize(data: dict) -> str: - return json.dumps(data, default=_json_default) - - -def _json_default(obj: object) -> str: - if isinstance(obj, datetime): - return obj.isoformat() - raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") +from source.models import SCHEMA_CONTRACT + +def init_lake_store(source_conn: duckdb.DuckDBPyConnection) -> duckdb.DuckDBPyConnection: + """Creates the structural schema for an immutable append-only historical audit ledger.""" + lake_conn = duckdb.connect(':memory:') + + for table_name in SCHEMA_CONTRACT.keys(): + source_info = source_conn.execute(f"PRAGMA table_info('{table_name}')").fetchall() + col_definitions = [f"{col[1]} {col[2]}" for col in source_info] + + # Inject auditing metadata fields required to guarantee perfect recovery tracking + col_definitions.extend([ + "_cdc_op_type CHAR(1)", + "_cdc_extracted_at TIMESTAMP", + "_cdc_lsn_version BIGINT" + ]) + + definitions_str = ", ".join(col_definitions) + lake_conn.execute(f"CREATE TABLE lake_{table_name}_history ({definitions_str});") + + return lake_conn \ No newline at end of file diff --git a/submission/sample-candidate/pipeline/warehouse.py b/submission/sample-candidate/pipeline/warehouse.py index ab32401..3cbbaed 100644 --- a/submission/sample-candidate/pipeline/warehouse.py +++ b/submission/sample-candidate/pipeline/warehouse.py @@ -1,124 +1,22 @@ -""" -Warehouse layer — current-state snapshot built from CDC events. - -Each warehouse table mirrors the source table with two extra columns: - _cdc_seq : sequence of the last CDC event that touched this row - _deleted : soft-delete flag set when a DELETE event is received - -Production analogue: BigQuery/Snowflake tables refreshed by a streaming -merge job keyed on primary key. SCD2 history tables would sit alongside -these current-state tables for time-travel queries. -""" - -from __future__ import annotations - -import duckdb - -from pipeline.cdc import CDCRecord - -# Source table → warehouse table -_TABLE_MAP: dict[str, str] = { - "customers": "wh_customers", - "wallets": "wh_wallets", - "transactions": "wh_transactions", -} - -# Source table → primary key column name -_PK_MAP: dict[str, str] = { - "customers": "customer_id", - "wallets": "wallet_id", - "transactions": "transaction_id", -} - - -def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_customers ( - customer_id VARCHAR PRIMARY KEY, - name VARCHAR, - email VARCHAR, - status VARCHAR, - created_at TIMESTAMP, - updated_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false - ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_wallets ( - wallet_id VARCHAR PRIMARY KEY, - customer_id VARCHAR, - balance DECIMAL(18, 2), - currency VARCHAR, - status VARCHAR, - created_at TIMESTAMP, - updated_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false +from source.models import SCHEMA_CONTRACT + +def project_warehouse_table(lake_conn, table_name: str, as_of_time) -> list: + """Runs a late-arriving resilient window partition to flatten historical logs down into snapshots.""" + cols_str = ", ".join(SCHEMA_CONTRACT[table_name]) + pk_col = "transaction_id" if table_name == "transactions" else f"{table_name[:-1]}_id" + + query = f""" + WITH partitioned_ledger AS ( + SELECT {cols_str}, _cdc_op_type, + ROW_NUMBER() OVER( + PARTITION BY {pk_col} + ORDER BY _cdc_lsn_version DESC, _cdc_extracted_at DESC + ) as tracking_rank + FROM lake_{table_name}_history + WHERE _cdc_extracted_at <= ? ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS wh_transactions ( - transaction_id VARCHAR PRIMARY KEY, - wallet_id VARCHAR, - amount DECIMAL(18, 2), - direction VARCHAR, - status VARCHAR, - reference VARCHAR, - created_at TIMESTAMP, - settled_at TIMESTAMP, - _cdc_seq INTEGER NOT NULL, - _deleted BOOLEAN NOT NULL DEFAULT false - ) - """) - - -def apply_cdc_records( - conn: duckdb.DuckDBPyConnection, records: list[CDCRecord] -) -> None: - """ - Apply CDC records to the warehouse current-state tables in sequence order. - - - insert / update → upsert (insert new row or overwrite existing) - - delete → set _deleted = true + SELECT {cols_str} + FROM partitioned_ledger + WHERE tracking_rank = 1 AND _cdc_op_type != 'D' """ - for record in sorted(records, key=lambda r: r.sequence): - wh_table = _TABLE_MAP.get(record.table) - pk_col = _PK_MAP.get(record.table) - if not wh_table or not pk_col: - continue - - pk_val = record.primary_key - - if record.operation == "delete": - conn.execute( - f"UPDATE {wh_table} SET _deleted = true, _cdc_seq = ?" - f" WHERE {pk_col} = ?", - [record.sequence, pk_val], - ) - continue - - data = {**record.data, "_cdc_seq": record.sequence, "_deleted": False} - cols = list(data.keys()) - vals = list(data.values()) - placeholders = ", ".join(["?"] * len(vals)) - - existing = conn.execute( - f"SELECT COUNT(*) FROM {wh_table} WHERE {pk_col} = ?", - [pk_val], - ).fetchone()[0] - - if existing: - set_clause = ", ".join([f"{c} = ?" for c in cols]) - conn.execute( - f"UPDATE {wh_table} SET {set_clause} WHERE {pk_col} = ?", - vals + [pk_val], - ) - else: - col_list = ", ".join(cols) - conn.execute( - f"INSERT INTO {wh_table} ({col_list}) VALUES ({placeholders})", - vals, - ) + return lake_conn.execute(query, (as_of_time,)).fetchall() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/check_schema_contracts.py b/submission/sample-candidate/scripts/check_schema_contracts.py index 1698b21..05a1ff1 100644 --- a/submission/sample-candidate/scripts/check_schema_contracts.py +++ b/submission/sample-candidate/scripts/check_schema_contracts.py @@ -1,60 +1,23 @@ -#!/usr/bin/env python3 -""" -check_schema_contracts.py - -Validates that the source tables expose all columns defined in the schema -contract. A missing column means downstream CDC logic or warehouse models -will break — this script fails the build before that happens. - -Exit 0 — all contracts pass. -Exit 1 — one or more violations found. -""" - -import os import sys - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - import duckdb - -from source.models import SCHEMA_CONTRACT, create_source_tables - - -def check_contracts(conn: duckdb.DuckDBPyConnection) -> list[str]: - """Return a list of violation messages. Empty list = all passed.""" - violations: list[str] = [] - - for table, expected_cols in SCHEMA_CONTRACT.items(): - try: - rows = conn.execute(f"DESCRIBE {table}").fetchall() - except Exception as exc: - violations.append(f"{table}: could not describe table — {exc}") - continue - - actual_cols = {row[0] for row in rows} - - for col in expected_cols: - if col not in actual_cols: - violations.append(f"{table}.{col}: column missing from source table") - - return violations - - -def main() -> int: - conn = duckdb.connect(":memory:") - create_source_tables(conn) - - violations = check_contracts(conn) - - if violations: - print("Schema contract violations:") - for v in violations: - print(f" ✗ {v}") - return 1 - - print(f"All schema contracts passed ({len(SCHEMA_CONTRACT)} tables checked).") - return 0 - +from source.models import create_source_tables +from pipeline.lake import init_lake_store +from pipeline.cdc import CDCPipeline, SchemaMismatchException + +def run_contract_verification(): + print("Executing pre-flight platform safety contract scans...") + src_db = duckdb.connect(':memory:') + create_source_tables(src_db) + lake_db = init_lake_store(src_db) + + pipeline = CDCPipeline(src_db, lake_db) + + try: + pipeline.verify_schema_contract() + print("Success: Source database validation is compliant with schemas.") + except SchemaMismatchException as e: + print(f"Verification Failure: {str(e)}") + sys.exit(1) if __name__ == "__main__": - sys.exit(main()) + run_contract_verification() \ No newline at end of file diff --git a/submission/sample-candidate/scripts/run_data_quality_checks.py b/submission/sample-candidate/scripts/run_data_quality_checks.py index 31e4cba..1021e0d 100644 --- a/submission/sample-candidate/scripts/run_data_quality_checks.py +++ b/submission/sample-candidate/scripts/run_data_quality_checks.py @@ -1,212 +1,18 @@ -#!/usr/bin/env python3 -""" -run_data_quality_checks.py - -Runs system and business data quality validations against the warehouse. -Seeds an in-memory database with representative data, applies CDC events, -then asserts correctness invariants. - -System checks : PK uniqueness, not-null, referential integrity -Business checks : non-negative balances, positive amounts, valid status enums - -Exit 0 — all checks pass. -Exit 1 — one or more failures found. -""" - -import os -import sys - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from datetime import datetime, timezone - import duckdb -from pipeline.cdc import CDCCapture -from pipeline.lake import append_to_lake, create_lake_table -from pipeline.warehouse import apply_cdc_records, create_warehouse_tables - - -def _now() -> datetime: - return datetime.now(timezone.utc) - - -def seed(conn: duckdb.DuckDBPyConnection, capture: CDCCapture) -> None: - """Populate lake + warehouse with representative test data.""" - ts = _now() - - capture.insert( - "customers", - "c1", - { - "customer_id": "c1", - "name": "Alice", - "email": "alice-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "customers", - "c2", - { - "customer_id": "c2", - "name": "Bob", - "email": "bob-test-user", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - - capture.insert( - "wallets", - "w1", - { - "wallet_id": "w1", - "customer_id": "c1", - "balance": 100.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - capture.insert( - "wallets", - "w2", - { - "wallet_id": "w2", - "customer_id": "c2", - "balance": 50.00, - "currency": "USD", - "status": "active", - "created_at": ts, - "updated_at": ts, - }, - ) - - capture.insert( - "transactions", - "t1", - { - "transaction_id": "t1", - "wallet_id": "w1", - "amount": 25.00, - "direction": "credit", - "status": "settled", - "reference": "ref-001", - "created_at": ts, - "settled_at": ts, - }, - ) - capture.insert( - "transactions", - "t2", - { - "transaction_id": "t2", - "wallet_id": "w2", - "amount": 10.00, - "direction": "debit", - "status": "settled", - "reference": "ref-002", - "created_at": ts, - "settled_at": ts, - }, - ) - - create_lake_table(conn) - create_warehouse_tables(conn) - records = capture.records_since(0) - append_to_lake(conn, records) - apply_cdc_records(conn, records) - - -def run_checks(conn: duckdb.DuckDBPyConnection) -> list[str]: - failures: list[str] = [] - - # ── system checks ───────────────────────────────────────────────────────── - - pk_map = { - "wh_customers": "customer_id", - "wh_wallets": "wallet_id", - "wh_transactions": "transaction_id", - } - for table, pk in pk_map.items(): - total = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] - distinct = conn.execute(f"SELECT COUNT(DISTINCT {pk}) FROM {table}").fetchone()[ - 0 - ] - if total != distinct: - failures.append( - f"{table}: PK not unique — {total} rows, {distinct} distinct {pk}" - ) - - not_null_checks = [ - ("wh_customers", "name"), - ("wh_customers", "email"), - ("wh_wallets", "customer_id"), - ("wh_wallets", "currency"), - ("wh_transactions", "wallet_id"), - ("wh_transactions", "amount"), - ("wh_transactions", "direction"), - ] - for table, col in not_null_checks: - nulls = conn.execute( - f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL" - ).fetchone()[0] - if nulls: - failures.append(f"{table}.{col}: {nulls} NULL value(s) found") - - # ── business checks ─────────────────────────────────────────────────────── - - neg_bal = conn.execute( - "SELECT COUNT(*) FROM wh_wallets WHERE balance < 0" - ).fetchone()[0] - if neg_bal: - failures.append(f"wh_wallets: {neg_bal} row(s) with negative balance") - - non_pos = conn.execute( - "SELECT COUNT(*) FROM wh_transactions WHERE amount <= 0" - ).fetchone()[0] - if non_pos: - failures.append(f"wh_transactions: {non_pos} row(s) with non-positive amount") - - bad_dir = conn.execute( - "SELECT COUNT(*) FROM wh_transactions" - " WHERE direction NOT IN ('credit', 'debit')" - ).fetchone()[0] - if bad_dir: - failures.append(f"wh_transactions: {bad_dir} row(s) with invalid direction") - - # ── lake completeness ───────────────────────────────────────────────────── - - lake_count = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] - if lake_count == 0: - failures.append("lake_cdc_events: no records found — lake appears empty") - - return failures - - -def main() -> int: - conn = duckdb.connect(":memory:") - capture = CDCCapture() - seed(conn, capture) - - failures = run_checks(conn) - - if failures: - print("Data quality failures:") - for f in failures: - print(f" ✗ {f}") - return 1 - - print( - f"All data quality checks passed ({len(run_checks.__code__.co_consts)} rules checked)." - ) - return 0 - - -if __name__ == "__main__": - sys.exit(main()) +def execute_quality_suite(lake_conn): + print("Auditing active business rule metrics across layers...") + + # 1. Validation Parity Rule: Confirm non-negative wallet amounts + negative_balances = lake_conn.execute(""" + SELECT COUNT(*) FROM lake_wallets_history WHERE balance < 0 + """).fetchone()[0] + assert negative_balances == 0, "CRITICAL ERROR: Detected negative currency amounts inside historical records." + + # 2. Invariant Check: Enforce valid system status domains + invalid_statuses = lake_conn.execute(""" + SELECT COUNT(*) FROM lake_customers_history WHERE status NOT IN ('active', 'suspended', 'inactive') + """).fetchone()[0] + assert invalid_statuses == 0, "CRITICAL ERROR: Malformed enum parameters leaked past ingestion walls." + + print("All active data quality assertions passed.") \ No newline at end of file diff --git a/submission/sample-candidate/scripts/validate_catalog.py b/submission/sample-candidate/scripts/validate_catalog.py index 6a02026..b9c0593 100644 --- a/submission/sample-candidate/scripts/validate_catalog.py +++ b/submission/sample-candidate/scripts/validate_catalog.py @@ -1,77 +1,25 @@ -#!/usr/bin/env python3 -""" -validate_catalog.py - -Verifies that catalog/catalog.json is present and contains a complete entry -for every required lake and warehouse dataset. - -Required fields per entry: name, layer, description, owner, schema, update_cadence - -Exit 0 — catalog is valid. -Exit 1 — missing file, missing datasets, or missing required fields. -""" - import json import os -import sys - -CATALOG_PATH = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), - "catalog", - "catalog.json", -) - -REQUIRED_DATASETS = [ - "lake_cdc_events", - "wh_customers", - "wh_wallets", - "wh_transactions", -] - -REQUIRED_FIELDS = ["name", "layer", "description", "owner", "schema", "update_cadence"] - - -def validate() -> list[str]: - violations: list[str] = [] - - if not os.path.exists(CATALOG_PATH): - violations.append(f"catalog.json not found at {CATALOG_PATH}") - return violations - - with open(CATALOG_PATH) as f: - try: - catalog = json.load(f) - except json.JSONDecodeError as exc: - violations.append(f"catalog.json is not valid JSON: {exc}") - return violations - - datasets = {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} - - for name in REQUIRED_DATASETS: - if name not in datasets: - violations.append(f"Missing dataset entry: {name}") - continue - - entry = datasets[name] - for field in REQUIRED_FIELDS: - if field not in entry or not entry[field]: - violations.append(f"{name}: missing or empty required field '{field}'") - - return violations - - -def main() -> int: - violations = validate() - - if violations: - print("Catalog validation failures:") - for v in violations: - print(f" ✗ {v}") - return 1 - - print(f"Catalog validation passed ({len(REQUIRED_DATASETS)} datasets verified).") - return 0 +def run_catalog_validation(): + catalog_path = "catalog/catalog.json" + print(f"📋 Structural discovery validation for: {catalog_path}") + + if not os.path.exists(catalog_path): + print(" Error: Structural mapping schema file missing.") + return False + + with open(catalog_path, 'r') as f: + metadata = json.load(f) + + required_layers = ["lake_historical_ledger", "warehouse_curated_snapshots"] + for layer in required_layers: + if layer not in metadata: + print(f"Error: Discoverability registration path '{layer}' is unregistered.") + return False + + print("Discovery boundaries are successfully validated.") + return True if __name__ == "__main__": - sys.exit(main()) + run_catalog_validation() \ No newline at end of file diff --git a/submission/sample-candidate/source/models.py b/submission/sample-candidate/source/models.py index 91c9d19..dab947c 100644 --- a/submission/sample-candidate/source/models.py +++ b/submission/sample-candidate/source/models.py @@ -1,84 +1,51 @@ -""" -Source schema for a payments/wallet system. - -Domain: customers own wallets; wallets have transactions. - -Strong entities : customers, wallets -Weak entities : transactions (lifecycle tied to wallet) - -Invariants: -- wallet.balance >= 0 -- transaction.amount > 0 -- status fields restricted to known enum values -- settled_at must be >= created_at when present -""" - import duckdb -# Expected columns per table — used by schema-contract checks. -SCHEMA_CONTRACT: dict[str, list[str]] = { +# Immutable source-of-truth registry to guard downstream architecture +SCHEMA_CONTRACT = { "customers": ["customer_id", "name", "email", "status", "created_at", "updated_at"], - "wallets": [ - "wallet_id", - "customer_id", - "balance", - "currency", - "status", - "created_at", - "updated_at", - ], - "transactions": [ - "transaction_id", - "wallet_id", - "amount", - "direction", - "status", - "reference", - "created_at", - "settled_at", - ], + "wallets": ["wallet_id", "customer_id", "balance", "currency", "status", "created_at", "updated_at"], + "transactions": ["transaction_id", "sender_wallet_id", "receiver_wallet_id", "amount", "status", "created_at"] } - -def create_source_tables(conn: duckdb.DuckDBPyConnection) -> None: - """Create source tables with constraints in the given DuckDB connection.""" +def create_source_tables(conn: duckdb.DuckDBPyConnection): + """Initializes a relational payments network domain with strict precision types.""" + # Strict Enum Constraints + conn.execute("CREATE TYPE account_status AS ENUM ('active', 'suspended', 'inactive');") + conn.execute("CREATE TYPE tx_status AS ENUM ('PENDING', 'SUCCESS', 'FAILED');") + + # Customers (Strong Entity) conn.execute(""" CREATE TABLE IF NOT EXISTS customers ( - customer_id VARCHAR PRIMARY KEY, - name VARCHAR NOT NULL, - email VARCHAR NOT NULL, - status VARCHAR NOT NULL - CHECK (status IN ('active', 'suspended', 'closed')), - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + email VARCHAR NOT NULL UNIQUE, + status account_status NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ); """) - + + # Wallets (Weak Entity dependent on Customers) conn.execute(""" CREATE TABLE IF NOT EXISTS wallets ( - wallet_id VARCHAR PRIMARY KEY, - customer_id VARCHAR NOT NULL REFERENCES customers(customer_id), - balance DECIMAL(18, 2) NOT NULL DEFAULT 0.00 - CHECK (balance >= 0), - currency VARCHAR NOT NULL, - status VARCHAR NOT NULL - CHECK (status IN ('active', 'frozen', 'closed')), - created_at TIMESTAMP NOT NULL, - updated_at TIMESTAMP NOT NULL - ) + wallet_id VARCHAR PRIMARY KEY, + customer_id VARCHAR REFERENCES customers(customer_id), + balance DECIMAL(18, 2) NOT NULL CHECK (balance >= 0), + currency VARCHAR(3) NOT NULL, + status account_status NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ); """) - + + # Transactions (Append-Only Event Ledger) conn.execute(""" CREATE TABLE IF NOT EXISTS transactions ( - transaction_id VARCHAR PRIMARY KEY, - wallet_id VARCHAR NOT NULL REFERENCES wallets(wallet_id), - amount DECIMAL(18, 2) NOT NULL CHECK (amount > 0), - direction VARCHAR NOT NULL - CHECK (direction IN ('credit', 'debit')), - status VARCHAR NOT NULL - CHECK (status IN ('pending', 'settled', 'failed', 'reversed')), - reference VARCHAR, - created_at TIMESTAMP NOT NULL, - settled_at TIMESTAMP - ) - """) + transaction_id VARCHAR PRIMARY KEY, + sender_wallet_id VARCHAR REFERENCES wallets(wallet_id), + receiver_wallet_id VARCHAR REFERENCES wallets(wallet_id), + amount DECIMAL(18, 2) NOT NULL CHECK (amount > 0), + status tx_status NOT NULL, + created_at TIMESTAMP NOT NULL + ); + """) \ No newline at end of file