testing github action

.github/workflows/data-pr-checks.yml

@@ -22,13 +22,13 @@ jobs:
           pip install ruff black sqlfluff
 
       - name: Ruff
-        run: ruff check .
+        run: ruff check submission/
 
       - name: Black
-        run: black --check .
+        run: black --check submission/
 
       - name: SQLFluff
-        run: sqlfluff lint .
+        run: sqlfluff lint submission/ || true
         continue-on-error: true
 
   test-pipeline:
@@ -44,11 +44,11 @@ jobs:
       - name: Install test dependencies
         run: |
           python -m pip install --upgrade pip
-          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
           pip install pytest
+          if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi
 
       - name: Run tests
-        run: pytest -q
+        run: pytest submission/ -q
 
   validate-models:
     name: validate-models
@@ -95,12 +95,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+          if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi
 
       - name: Run schema contract checks
         run: |
-          if [ -f scripts/check_schema_contracts.py ]; then
-            python scripts/check_schema_contracts.py
+          if [ -f submission/scripts/check_schema_contracts.py ]; then
+            python submission/scripts/check_schema_contracts.py
           else
             echo "No schema contract script found. Add one or replace this step."
             exit 1
@@ -119,12 +119,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+          if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi
 
       - name: Run data quality checks
         run: |
-          if [ -f scripts/run_data_quality_checks.py ]; then
-            python scripts/run_data_quality_checks.py
+          if [ -f submission/scripts/run_data_quality_checks.py ]; then
+            python submission/scripts/run_data_quality_checks.py
           else
             echo "No data quality script found. Add one or replace this step."
             exit 1
@@ -143,12 +143,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+          if [ -f submission/requirements.txt ]; then pip install -r submission/requirements.txt; fi
 
       - name: Validate catalog metadata
         run: |
-          if [ -f scripts/validate_catalog.py ]; then
-            python scripts/validate_catalog.py
+          if [ -f submission/scripts/validate_catalog.py ]; then
+            python submission/scripts/validate_catalog.py
           else
             echo "No catalog validation script found. Add one or replace this step."
             exit 1
@@ -159,6 +159,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
 
       - uses: actions/setup-python@v5
         with:
@@ -179,3 +181,4 @@ jobs:
         uses: gitleaks/gitleaks-action@v2
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITLEAKS_LICENSE: ${{ secrets.GITLEAKS_LICENSE }}

submission/catalog/catalog.json

@@ -0,0 +1,77 @@
+{
+  "datasets": [
+    {
+      "name": "lake_cdc_events",
+      "layer": "lake",
+      "description": "Append-only log of every CDC event captured from the payments source system. Preserves full change history for replay and point-in-time recovery.",
+      "owner": "data-platform",
+      "consumers": ["data-platform", "analytics", "audit"],
+      "update_cadence": "real-time",
+      "schema": {
+        "sequence": "INTEGER — monotonic capture offset",
+        "operation": "VARCHAR — insert | update | delete",
+        "table_name": "VARCHAR — source table name",
+        "primary_key": "VARCHAR — source row PK value",
+        "data": "VARCHAR (JSON) — full row snapshot at capture time",
+        "captured_at": "TIMESTAMP — UTC capture time"
+      }
+    },
+    {
+      "name": "wh_customers",
+      "layer": "warehouse",
+      "description": "Current-state customer snapshot. Reflects the latest known state of each customer row. Soft-deleted rows are flagged with _deleted=true.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "product", "finance"],
+      "update_cadence": "near-real-time",
+      "schema": {
+        "customer_id": "VARCHAR — primary key",
+        "name": "VARCHAR",
+        "email": "VARCHAR",
+        "status": "VARCHAR — active | suspended | closed",
+        "created_at": "TIMESTAMP",
+        "updated_at": "TIMESTAMP",
+        "_cdc_seq": "INTEGER — sequence of last CDC event",
+        "_deleted": "BOOLEAN — true if source row was deleted"
+      }
+    },
+    {
+      "name": "wh_wallets",
+      "layer": "warehouse",
+      "description": "Current-state wallet snapshot. Balance reflects the latest value written via CDC. Negative balances are a data quality violation.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "finance"],
+      "update_cadence": "near-real-time",
+      "schema": {
+        "wallet_id": "VARCHAR — primary key",
+        "customer_id": "VARCHAR — FK to wh_customers",
+        "balance": "DECIMAL(18,2)",
+        "currency": "VARCHAR",
+        "status": "VARCHAR — active | frozen | closed",
+        "created_at": "TIMESTAMP",
+        "updated_at": "TIMESTAMP",
+        "_cdc_seq": "INTEGER",
+        "_deleted": "BOOLEAN"
+      }
+    },
+    {
+      "name": "wh_transactions",
+      "layer": "warehouse",
+      "description": "Current-state transaction snapshot. Each row is the latest state of a transaction from the source. Amount must always be positive.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "finance", "audit"],
+      "update_cadence": "near-real-time",
+      "schema": {
+        "transaction_id": "VARCHAR — primary key",
+        "wallet_id": "VARCHAR — FK to wh_wallets",
+        "amount": "DECIMAL(18,2) — always > 0",
+        "direction": "VARCHAR — credit | debit",
+        "status": "VARCHAR — pending | settled | failed | reversed",
+        "reference": "VARCHAR — nullable external reference",
+        "created_at": "TIMESTAMP",
+        "settled_at": "TIMESTAMP — nullable until settled",
+        "_cdc_seq": "INTEGER",
+        "_deleted": "BOOLEAN"
+      }
+    }
+  ]
+}

submission/conftest.py

@@ -0,0 +1,6 @@
+"""Root conftest — adds submission/ to sys.path so tests can import source/pipeline."""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))

submission/pipeline/cdc.py

@@ -0,0 +1,89 @@
+"""
+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

submission/pipeline/lake.py

@@ -0,0 +1,69 @@
+"""
+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")