Experimentation System: Schema Contracts

Build a CI drift gate that catches schema changes across 17 Delta tables.

Introduction

30 Second Summary

A column gets renamed. A new nullable field appears. A lookup table gains a value nobody told your code about. In data systems with many tables, schema drift is the silent killer that breaks pipelines hours or days after the change lands.

In this project, you will build a schema contract layer for a 17-table Marketing Experimentation System. You will auto-generate Pydantic models from live table metadata, introspect enums from lookup tables, and wire a two-directional drift gate into CI that fails the build on any mismatch.

What You'll Build

When you finish, you will run your CI pipeline and watch it reject an injected fake field in under a second. Any schema change to any of your 17 MES tables that is not reflected in the contract will fail before it reaches production.

By the end of this project, you'll have:

  • A schemas_generator.py that introspects all 17 tables from Unity Catalog and writes a complete schemas.py with Pydantic models and live enums pulled directly from your lookup tables.
  • A schema_validator.py drift gate that performs two-directional validation (model vs table, table vs model) plus enum drift detection, exiting with code 1 on any mismatch.
  • A proven failure-injection test demonstrating the drift gate catches regressions before they reach production, wired into your existing CI pipeline.
  • Secret Mission: Extend the validator with a schema evolution report that outputs a markdown diff of what changed since the last clean run.

Are there any prerequisites?

This project assumes you have an existing MES deployment with 17 Delta tables in marketing_experimentation.mes_dev, seeded lu_* lookup tables, a validated stats_engine.py with CI, and DATABRICKS_WAREHOUSE_ID configured in your environment.

You also need Python 3.10+ and databricks-sdk already installed with access to a running SQL warehouse.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters for your experimentation system.

Your 17-table MES has no mechanism to catch schema drift between the live Delta tables and the code that reads and writes them. A column rename, a new nullable field, or a lookup value addition can silently break pipelines. This project closes that gap.

Set Up Introspection

You committed to building a schema contract layer for your 17 MES tables. The first question is: where does the truth about those tables actually live?

Any contract you write by hand is stale the moment you type it. In this step, you'll build a generator script that reads metadata directly from the live data layer via Unity Catalog's information_schema. By the end, you'll have confirmed it sees all 17 tables.

In this step, get ready to:
  • Install Pydantic v2 and confirm databricks-sdk is available.
  • Create schemas_generator.py with live introspection from information_schema.
  • Build a SQL-to-Python type map and utility functions for the generator.
Install Pydantic v2 and confirm databricks-sdk

Your schema contracts will be expressed as Pydantic models. Pydantic v2 provides strict validation, ConfigDict, and field_validator decorators you'll use later.

  • Install Pydantic 2.13.4 by running this command:
pip install pydantic==2.13.4
  • Confirm Pydantic installed correctly by running:
python -c "import pydantic; print(pydantic.__version__)"

You should see 2.13.4 printed.

  • Confirm databricks-sdk is available by running:
python -c "from databricks.sdk import WorkspaceClient; print('OK')"

You should see OK printed.

Seeing ModuleNotFoundError?

If Pydantic fails to import, confirm you're using the same Python environment where pip installed it. Run which python (macOS/Linux) or where python (Windows) and compare to the interpreter your shell is using.

If databricks-sdk fails, confirm your existing environment has it installed. Run pip show databricks-sdk to check.

Help me debug my Python environment for Pydantic and databricks-sdk.

Build schemas_generator.py with the introspection query

The generator script will query information_schema.columns in your Unity Catalog to retrieve column metadata for every table in mes_dev. This is the live source of truth for your schema contracts.

  • Create a new file called schemas_generator.py and add the module header, imports, constants, and the execute_sql() helper:
"""
MES Schema Generator
Introspects marketing_experimentation.mes_dev and generates Pydantic models + enums.
"""
import os
import re
from databricks.sdk import WorkspaceClient

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"
WAREHOUSE_ID = os.environ["DATABRICKS_WAREHOUSE_ID"]


def execute_sql(w: WorkspaceClient, sql: str) -> list[dict]:
    """Execute SQL via Statement Execution API and return rows as dicts."""
    response = w.statement_execution.execute_statement(
        warehouse_id=WAREHOUSE_ID,
        statement=sql,
        wait_timeout="50s",
    )
    if response.status.state.value == "FAILED":
        raise RuntimeError(f"SQL failed: {response.status.error.message}")
    columns = [col.name for col in response.manifest.schema.columns]
    rows = []
    if response.result and response.result.data_array:
        for row_data in response.result.data_array:
            rows.append(dict(zip(columns, row_data)))
    return rows

What does this code do?

  • CATALOG, SCHEMA, and WAREHOUSE_ID point the script at your live MES deployment and the SQL warehouse that runs queries.
  • execute_sql() wraps the Statement Execution API from databricks-sdk. It sends a SQL string to your warehouse, waits up to 50 seconds, and returns the result rows as a list of dictionaries.
  • The error handling raises immediately if the query fails, so you get a clear traceback instead of silent None values.
  • Now add the get_all_tables() function and the main() entry point below execute_sql():
def get_all_tables(w: WorkspaceClient) -> dict[str, list[dict]]:
    """Query information_schema for all tables in mes_dev, return grouped columns."""
    sql = f"""
    SELECT table_name, column_name, data_type, is_nullable
    FROM {CATALOG}.information_schema.columns
    WHERE table_catalog = '{CATALOG}'
      AND table_schema = '{SCHEMA}'
    ORDER BY table_name, ordinal_position
    """
    rows = execute_sql(w, sql)
    tables: dict[str, list[dict]] = {}
    for row in rows:
        table = row["table_name"]
        if table not in tables:
            tables[table] = []
        tables[table].append(row)
    return tables


def main():
    w = WorkspaceClient()
    tables = get_all_tables(w)
    print(f"Found {len(tables)} tables:")
    for name, cols in sorted(tables.items()):
        print(f"  {name}: {len(cols)} columns")
    if len(tables) != 17:
        print(f"WARNING: Expected 17 tables, found {len(tables)}")


if __name__ == "__main__":
    main()

What does this code do?

  • get_all_tables() queries information_schema.columns filtering by your catalog and schema. It groups the result rows by table name so each table maps to its list of column metadata.
  • main() is a dry-run that prints each table name and column count. The warning at the end catches unexpected table counts early.
  • Unity Catalog stores identifiers as lowercase, so the WHERE filters compare directly with lowercase string literals.
  • Save schemas_generator.py.

Before you run this, how many tables do you expect to see printed?

  • Run the generator in dry-run mode:
python schemas_generator.py

You should see exactly 17 table names, each with its column count. The output starts with Found 17 tables: followed by an indented list.

Not seeing 17 tables?

If you see 0 tables, confirm your DATABRICKS_WAREHOUSE_ID environment variable is set and the warehouse is running. Run echo $DATABRICKS_WAREHOUSE_ID to check.

If you get an authentication error, confirm your Databricks credentials are configured (token or OAuth via environment variables or .databrickscfg).

If you see a different table count, verify you have the full MES deployment in marketing_experimentation.mes_dev.

Help me debug why schemas_generator.py is not finding my 17 MES tables.

Build the SQL-to-Python type map and utilities

When the generator builds Pydantic models, it needs to convert Databricks SQL types like DECIMAL(10,2) into Python types like float. You'll add the mapping dictionary and two utility functions now.

  • In schemas_generator.py, find the line WAREHOUSE_ID = os.environ["DATABRICKS_WAREHOUSE_ID"].
  • Add the following two dictionaries directly below it (above the execute_sql function):
SQL_TYPE_MAP = {
    "STRING": "str",
    "INT": "int",
    "BIGINT": "int",
    "DOUBLE": "float",
    "FLOAT": "float",
    "DECIMAL": "float",
    "BOOLEAN": "bool",
    "DATE": "date",
    "TIMESTAMP": "datetime",
    "ARRAY": "list",
    "MAP": "dict",
    "STRUCT": "dict",
}

ENUM_TABLES = {
    "lu_statuses": "ExperimentStatus",
    "lu_methods": "ExperimentMethod",
    "lu_assignment_units": "AssignmentUnit",
    "lu_decision_outcomes": "DecisionOutcome",
    "lu_metric_types": "MetricType",
    "lu_readout_types": "ReadoutType",
}

What do these dictionaries do?

SQL_TYPE_MAP maps every Databricks SQL base type to its Python equivalent. Parameterized types like DECIMAL(10,2) or ARRAY<STRING> will be stripped down to their base (DECIMAL, ARRAY) before lookup.

ENUM_TABLES maps each lu_* lookup table name to the Python enum class name the generator will create. These six tables hold all valid values for experiment statuses, methods, and other categorical fields.

  • Now add the two utility functions between ENUM_TABLES and execute_sql():
def map_sql_type(sql_type: str) -> str:
    """Map a Databricks SQL type string to a Python type string."""
    base_type = re.split(r"[<(]", sql_type.upper())[0].strip()
    return SQL_TYPE_MAP.get(base_type, "str")


def table_name_to_class_name(name: str) -> str:
    """Convert snake_case table name to PascalCase class name."""
    return "".join(word.capitalize() for word in name.split("_"))

What do these functions do?

  • map_sql_type() strips parameterized type syntax (everything after < or () and looks up the base type in SQL_TYPE_MAP. Unknown types default to str.
  • table_name_to_class_name() turns experiment_metrics into ExperimentMetrics so each model class gets a proper PascalCase name.
  • Save schemas_generator.py.
  • Verify the utilities work by running:
python -c "from schemas_generator import map_sql_type, table_name_to_class_name; print(map_sql_type('DECIMAL(10,2)')); print(table_name_to_class_name('experiment_metrics'))"

You should see float on the first line and ExperimentMetrics on the second.

Seeing a SyntaxError or ImportError?

Check that SQL_TYPE_MAP and ENUM_TABLES are placed AFTER the WAREHOUSE_ID line and BEFORE the def execute_sql line. The two utility functions should appear between ENUM_TABLES and execute_sql.

Confirm there is a blank line between each function definition. Missing blank lines between top-level definitions can cause import issues.

Help me fix a syntax or import error in schemas_generator.py.

✔️ Awesome, I've got everything!

Great. Confirm you've saved schemas_generator.py and the dry-run still prints 17 tables if you re-run it.

ⓧ I'd like to double check the full code

Your schemas_generator.py should contain these sections, in order:

  • Module docstring, import os, import re, from databricks.sdk import WorkspaceClient.
  • CATALOG, SCHEMA, WAREHOUSE_ID constants.
  • SQL_TYPE_MAP dictionary (12 entries mapping SQL types to Python types).
  • ENUM_TABLES dictionary (6 entries mapping lu_* table names to enum class names).
  • map_sql_type() function using re.split to strip parameterized types.
  • table_name_to_class_name() function splitting on underscores and capitalizing.
  • execute_sql() function using the Statement Execution API.
  • get_all_tables() function querying information_schema.columns.
  • main() function printing table names and column counts.
  • if __name__ == "__main__": main() guard at the bottom.

Scroll up and confirm each section is present in your file. Re-run python schemas_generator.py to confirm it still prints 17 tables.

Your introspection foundation is in place. The generator can see all 17 tables and knows how to map their SQL types to Python. Next, you'll make it introspect the lookup tables so your enums stay in sync with live data.

Introspect Enums from Lookup Tables

Your introspection foundation is solid. You can pull metadata for all 17 tables from information_schema. Now there is a second source of truth hiding in the data layer: the lookup tables.

The six lu_* tables define every valid value for status, method, assignment unit, decision outcome, metric type, and readout type. Hardcoding those values into Python enums creates a second contract that silently drifts the moment someone adds a new lookup value. This step makes enums introspected from the live tables, not invented.

In this step, get ready to:
  • Implement get_enum_definitions() to query each lookup table for its distinct values.
  • Verify full-set equality between live lookup values and generated enum members for all six tables.
  • Wire enum code generation into the pipeline with generate_enum_code().
Implement get_enum_definitions()

Each lookup table has a value column containing the canonical set of allowed values. Your function will query each table and collect those values into a dictionary keyed by enum class name.

  • In schemas_generator.py, find the get_all_tables() function.
  • Add the following get_enum_definitions() function directly below it:
def get_enum_definitions(w: WorkspaceClient) -> dict[str, list[str]]:
    """
    Query each lu_* table and return enum class definitions.

    MANDATORY RULE: The seeded lu_* lookup tables are the single source of truth
    for every enum. Enums are introspected, never hardcoded -- a hardcoded enum
    list is a second contract that will drift.
    """
    enums: dict[str, list[str]] = {}
    for table_name, class_name in ENUM_TABLES.items():
        sql = f"""
        SELECT DISTINCT value
        FROM {CATALOG}.{SCHEMA}.{table_name}
        ORDER BY value
        """
        rows = execute_sql(w, sql)
        values = [row["value"] for row in rows if row["value"]]
        enums[class_name] = values
    return enums

What does this code do?

The function iterates over the ENUM_TABLES dictionary you defined in Step 1. For each lookup table, it runs a SELECT DISTINCT value query to get the full set of allowed values directly from the live data.

The docstring contains the mandatory rule for this project: enums come from the lookup tables, never from hardcoded lists. A hardcoded list is a second contract that will drift.

Why introspect instead of hardcode?

Imagine someone adds a new experiment method to lu_methods. If your enum is hardcoded, the pipeline silently accepts the old set and rejects any record with the new method. Introspection means the generated enum always matches what the data layer actually allows.

  • Save schemas_generator.py.
Verify full-set equality against live lookup tables

A spot-check of individual values is not enough. You need bidirectional set equality: every live value appears in the enum, and no enum member exists that is absent from the live table.

  • In schemas_generator.py, find the main() function. Find this line:
    if len(tables) != 17:
        print(f"WARNING: Expected 17 tables, found {len(tables)}")
  • Replace those two lines with the following updated main() tail that adds an early return on mismatch, then calls your new function:
    if len(tables) != 17:
        print(f"WARNING: Expected 17 tables, found {len(tables)}")
        return
    enums = get_enum_definitions(w)
    print(f"\nIntrospected {len(enums)} enum tables")
    for class_name, values in sorted(enums.items()):
        print(f"  {class_name}: {len(values)} values")

What changed in main()?

  • Added return after the 17-table warning so the script stops early if the table count is wrong.
  • Called get_enum_definitions(w) and printed each enum class name with its value count.
  • Save schemas_generator.py.

Before you run this, what do you expect to see printed for the six enum tables? How many values do you think each lookup table contains?

  • Run the generator script to introspect the enum values:
python schemas_generator.py

You should see the 17 tables printed as before, followed by a section showing 6 introspected enum tables with value counts for each.

Now confirm full-set equality. For each lookup table, you will compare the SQL output directly against what the script reported.

  • Run this verification query in your Databricks SQL editor or via the SDK to check one table at a time. Start with statuses:
SELECT DISTINCT value
FROM marketing_experimentation.mes_dev.lu_statuses
ORDER BY value
  • Confirm the result set includes paused and archived statuses.
  • Run the same query pattern for each remaining table:
SELECT DISTINCT value FROM marketing_experimentation.mes_dev.lu_methods ORDER BY value;
SELECT DISTINCT value FROM marketing_experimentation.mes_dev.lu_assignment_units ORDER BY value;
SELECT DISTINCT value FROM marketing_experimentation.mes_dev.lu_decision_outcomes ORDER BY value;
SELECT DISTINCT value FROM marketing_experimentation.mes_dev.lu_metric_types ORDER BY value;
SELECT DISTINCT value FROM marketing_experimentation.mes_dev.lu_readout_types ORDER BY value;
  • For each table, verify bidirectional equality: every value in the SQL result appears in the script's output, and no extra values exist in the script's output that are absent from the SQL result.
  • Confirm these specific values are present across the six tables: aa_test in methods, time_block in assignment units, scale_pending_confirmation in decision outcomes, and interim/final/post_hoc/confirmation in readout types.

Why full-set equality instead of spot-checks?

A spot-check proves one value exists. Full-set equality proves the enum is neither missing values nor carrying stale ones. Both directions matter: a missing value means valid data gets rejected, and an extra value means the contract is looser than reality.

Seeing fewer values than expected?

Make sure your lu_* tables are seeded. If any lookup table returns zero rows, the seeding step from your MES deployment was incomplete. Re-run the seeding script for that table before continuing.

Also confirm your DATABRICKS_WAREHOUSE_ID env var points to a running SQL warehouse. A stopped warehouse causes timeouts, not empty results.

Help me debug why my lookup table query returns fewer values than expected.

Wire enum generation into the generator pipeline

You have the raw values. Now you need a function that turns them into Python Enum class source code. Each enum class inherits from both str and Enum so values are directly comparable to strings.

  • In schemas_generator.py, add the following generate_enum_code() function below get_enum_definitions():
def generate_enum_code(enums: dict[str, list[str]]) -> str:
    """Generate Python enum class source code from introspected values."""
    lines = []
    for class_name, values in sorted(enums.items()):
        lines.append(f"class {class_name}(str, Enum):")
        for val in values:
            member_name = re.sub(r"[^a-zA-Z0-9_]", "_", val).upper()
            lines.append(f'    {member_name} = "{val}"')
        lines.append("")
    return "\n".join(lines)

What does this code do?

  • For each enum class, it writes a class definition inheriting from str, Enum.
  • Each value becomes an uppercased member name. The regex replaces non-alphanumeric characters with underscores so names like aa_test become AA_TEST.
  • The original value string is preserved as the member's value, keeping string comparisons with raw data intact.
  • Save schemas_generator.py.
  • Run the script one more time to confirm everything still executes cleanly:
python schemas_generator.py

You should see the same 17 tables and 6 enum tables output as before. No errors means the new function is syntactically correct and ready for the model generation step.

Seeing a NameError or IndentationError?

Check that generate_enum_code() is at the module level (not indented inside another function). Also confirm that import re is still at the top of your file from Step 1.

Help me fix a syntax or import error in my schemas_generator.py after adding generate_enum_code.

✔️ Awesome, I've got everything!

Confirm you have saved schemas_generator.py with both get_enum_definitions() and generate_enum_code() defined, and the updated main() that calls get_enum_definitions().

ⓧ I'd like to double check the full code

Scroll up and confirm these pieces are all present in schemas_generator.py, in this order:

  • Module docstring, imports (os, re, WorkspaceClient), constants (CATALOG, SCHEMA, WAREHOUSE_ID), SQL_TYPE_MAP dict, ENUM_TABLES dict.
  • execute_sql() using the Statement Execution API.
  • map_sql_type() and table_name_to_class_name() utilities.
  • get_all_tables() querying information_schema.columns.
  • get_enum_definitions() with the MANDATORY RULE docstring, iterating ENUM_TABLES and querying SELECT DISTINCT value.
  • generate_enum_code() producing class Name(str, Enum) source code.
  • main() calling get_all_tables(), checking for 17 tables (with return on mismatch), then calling get_enum_definitions() and printing results.

Your enums are now introspected from live data, never hardcoded. Next up, you will turn all 17 tables into strict Pydantic models with B2B field validators that reject invalid experiment parameters at the boundary.

Generate Pydantic Models

Your generator can now introspect every column across all 17 tables and pull live enum values from each lookup table. That is raw metadata. Without a strict validation layer on top, there is nothing to enforce against.

In this step, you turn that metadata into Pydantic models that ARE the contract. Each model maps one-to-one with a Delta table, rejects unexpected fields, and enforces B2B-specific business rules on experiment design parameters.

In this step, get ready to:
  • Implement generate_model_code() that turns column metadata into a Pydantic BaseModel per table.
  • Add B2B field validators enforcing mde_pct, alpha, power, and split_ratio constraints.
  • Assemble and write the complete schemas.py file with a DO-NOT-EDIT banner, enums, and all 17 models.
Implement generate_model_code()

Each of your 17 tables needs a Pydantic BaseModel where nullable columns become Optional fields with a default of None, non-nullable columns are required, and any extra field triggers a validation error.

  • In schemas_generator.py, find the generate_enum_code() function.
  • Add the following function below it:
def generate_model_code(table_name: str, columns: list[dict]) -> str:
    """Generate a Pydantic model class for a single table."""
    class_name = table_name_to_class_name(table_name)
    lines = [
        f"class {class_name}(BaseModel):",
        "    model_config = ConfigDict(extra='forbid')",
        "",
    ]
    for col in columns:
        python_type = map_sql_type(col["data_type"])
        field_name = col["column_name"]
        if col["is_nullable"] == "YES":
            lines.append(f"    {field_name}: Optional[{python_type}] = None")
        else:
            lines.append(f"    {field_name}: {python_type}")
    return "\n".join(lines)

What does this code do?

  • table_name_to_class_name converts the snake_case table name (e.g. experiment_metrics) into PascalCase (ExperimentMetrics) for the class name.
  • ConfigDict(extra='forbid') makes any field not declared in the model raise a ValidationError. This is what catches stale models that have fewer fields than the live table.
  • The nullable check uses the IS_NULLABLE value from information_schema (returns 'YES' or 'NO' as strings) to decide whether a field is Optional or required.
  • Save schemas_generator.py.
Add B2B field validators

Experiment design parameters have hard business constraints. A minimum detectable effect below 5% wastes traffic on underpowered tests. An alpha above 10% makes results unreliable. These guardrails need to live in the contract, not in a wiki.

  • Below generate_model_code(), add this function that returns the validator source code as a string:
def generate_validators_code() -> str:
    """Generate B2B field validators for experiment design parameters."""
    return '''
    @field_validator("mde_pct")
    @classmethod
    def validate_mde_pct(cls, v):
        if v is not None and not (0.05 <= v <= 0.50):
            raise ValueError("mde_pct must be between 0.05 and 0.50")
        return v

    @field_validator("alpha")
    @classmethod
    def validate_alpha(cls, v):
        if v is not None and not (0.01 <= v <= 0.10):
            raise ValueError("alpha must be between 0.01 and 0.10")
        return v

    @field_validator("power")
    @classmethod
    def validate_power(cls, v):
        if v is not None and not (0.70 <= v <= 0.99):
            raise ValueError("power must be between 0.70 and 0.99")
        return v

    @field_validator("split_ratio")
    @classmethod
    def validate_split_ratio(cls, v):
        if v is not None and not (0.10 <= v <= 0.90):
            raise ValueError("split_ratio must be between 0.10 and 0.90")
        return v
'''

What does this code do?

  • Each @field_validator fires when a value is assigned to the named field on any model that has it.
  • mde_pct (minimum detectable effect) is constrained to 0.05-0.50. Below 5% requires unrealistic sample sizes for B2B traffic volumes.
  • alpha (0.01-0.10), power (0.70-0.99), and split_ratio (0.10-0.90) enforce statistical validity guardrails.
  • The function returns source code as a string because the generator writes it into schemas.py. It is not executed at generation time.
  • Save schemas_generator.py.
  • Confirm there are no syntax errors by running:
python -c "import schemas_generator; print('OK')"

You should see OK printed with no errors.

Seeing a SyntaxError?

  • Check that the triple-quoted string in generate_validators_code uses three single quotes (''') at both the opening and closing.
  • Confirm no stray indentation outside the function body. The return statement should be at 4-space indent inside the function.
  • help me debug a SyntaxError in my schemas_generator.py
Assemble and write schemas.py

The individual pieces are ready: model generation, enum generation, and validators. Now you need an orchestrator that assembles them into a single output file with the correct structure.

  • Below generate_validators_code(), add this function:
def generate_schemas_file(tables: dict[str, list[dict]], enums: dict[str, list[str]]) -> str:
    """Assemble the complete schemas.py content."""
    banner = '"""DO NOT EDIT - Auto-generated by schemas_generator.py"""\n'
    imports = (
        "from __future__ import annotations\n"
        "from datetime import date, datetime\n"
        "from enum import Enum\n"
        "from typing import Optional\n\n"
        "from pydantic import BaseModel, ConfigDict, field_validator\n\n"
    )
    enum_code = generate_enum_code(enums)
    model_sections = []
    validator_fields = {"mde_pct", "alpha", "power", "split_ratio"}
    for table_name in sorted(tables.keys()):
        columns = tables[table_name]
        model_code = generate_model_code(table_name, columns)
        table_fields = {col["column_name"] for col in columns}
        if validator_fields & table_fields:
            model_code += "\n" + generate_validators_code()
        model_sections.append(model_code)
    return banner + imports + enum_code + "\n\n" + "\n\n\n".join(model_sections) + "\n"

What does this code do?

  • The DO-NOT-EDIT banner warns anyone opening schemas.py that manual edits will be overwritten on the next generation run.
  • Models are sorted alphabetically by table name so diffs stay stable across regenerations.
  • The validator_fields set intersection check attaches validators only to models that actually contain those fields. Not every table has mde_pct or alpha.

Now update main() to call the assembler and write the file.

  • In schemas_generator.py, scroll to the main() function.
  • Find these lines at the end of main():
    for class_name, values in sorted(enums.items()):
        print(f"  {class_name}: {len(values)} values")
  • Add the following lines directly below them, at the same indentation level:
    content = generate_schemas_file(tables, enums)
    with open("schemas.py", "w") as f:
        f.write(content)
    print("\nGenerated schemas.py successfully.")
  • Save schemas_generator.py.
  • Run the generator to produce schemas.py:
python schemas_generator.py

You should see all 17 tables listed with their column counts, the 6 enum tables with their value counts, and Generated schemas.py successfully. at the end.

  • Confirm all 17 models import cleanly by running:
python -c "import schemas; print('OK')"

You should see OK with no import errors.

Import fails or missing module error?

  • If you see ModuleNotFoundError: No module named 'pydantic', confirm Pydantic is installed with pip install pydantic==2.13.4.
  • If you see a SyntaxError in schemas.py, re-run the generator. The generated file should never be hand-edited.
  • help me debug my schemas.py import error

Now for the real test. Before you run the next command, predict: what do you think happens when you try to create an Experiment with mde_pct=0.03? Will it succeed or raise an error?

  • Test the B2B validator by running:
python -c "from schemas import Experiment; Experiment(mde_pct=0.03)"

You should see a ValidationError stating that mde_pct must be between 0.05 and 0.50. The value 0.03 is below the minimum, so the contract rejects it.

Why does this matter?

Without this validator, a pipeline could silently insert an experiment with mde_pct=0.03 into your Delta table. That experiment would be statistically underpowered for B2B traffic volumes, waste budget, and produce inconclusive results. The contract catches it at write time.

No ValidationError raised?

  • Check that your Experiment model's table actually has a mde_pct column. If it does not, the validator will not be attached to that model.
  • Confirm the validator_fields set in generate_schemas_file() includes mde_pct and that the intersection check uses & (set intersection).
  • help me figure out why my Pydantic validator isn't firing

✔️ Awesome, I've got everything!

Great. Confirm you have saved schemas_generator.py and that the generated schemas.py exists in the same directory.

ⓧ I'd like to double check the full code

Scroll up and confirm each of these pieces is present in your schemas_generator.py, in order:

  • The existing functions from previous steps: execute_sql, map_sql_type, table_name_to_class_name, get_all_tables, get_enum_definitions, generate_enum_code.
  • generate_model_code() that builds a BaseModel class per table with ConfigDict(extra='forbid').
  • generate_validators_code() returning the four @field_validator blocks as a string.
  • generate_schemas_file() assembling banner + imports + enums + sorted models.
  • main() calling generate_schemas_file() and writing to schemas.py.

Your 17 models are generated, your enums are introspected, and your B2B validators fire on invalid design parameters. Next up, you build the CI drift gate that runs these contracts against the live tables on every commit.

Build the CI Drift Gate

Your generated schemas.py now contains 17 Pydantic models with strict validation rules. That contract is real, but nobody is checking it.

Without a validator that runs in CI, schema drift is invisible. A renamed column, a new required field, or a stale enum can silently break pipelines for days before anyone notices. This step builds the enforcement mechanism that makes drift a hard failure.

In this step, get ready to:
  • Build schema_validator.py with forward and reverse table checks.
  • Add enum drift detection across all six lookup tables.
  • Wire the validator into CI and prove it catches drift with a fake-field injection.
Build schema_validator.py with two-directional checks

The validator needs to compare your Pydantic models against the live table schemas in two directions. A forward check catches fields in the model that no longer exist in the table. A reverse check catches new non-nullable columns in the table that the model does not know about.

  • Create a new file called schema_validator.py in the same directory as schemas_generator.py.
  • Add the module setup with imports, constants, and the reverse type map by pasting this code:
"""
MES Schema Validator (CI Drift Gate)
Two-directional schema validation + enum drift detection.
Exit 0 = all pass, Exit 1 = any failure.
"""
import inspect
import os
import re
import sys
from enum import Enum

from databricks.sdk import WorkspaceClient

import schemas

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"
WAREHOUSE_ID = os.environ["DATABRICKS_WAREHOUSE_ID"]

# Reverse type map: Python type -> compatible SQL types
SQL_TYPE_MAP = {
    "str": ["STRING"],
    "int": ["INT", "BIGINT", "INTEGER", "SMALLINT", "TINYINT"],
    "float": ["DOUBLE", "FLOAT", "DECIMAL"],
    "bool": ["BOOLEAN"],
    "date": ["DATE"],
    "datetime": ["TIMESTAMP"],
    "list": ["ARRAY"],
    "dict": ["MAP", "STRUCT"],
}

What does this code do?

This sets up the reverse of the type map you built in schemas_generator.py. Instead of SQL-to-Python, this maps Python types back to all compatible SQL types.

For example, Python int is compatible with SQL INT, BIGINT, INTEGER, SMALLINT, and TINYINT. The validator uses this to check type compatibility in both directions.

  • Add the enum table mapping and the SQL execution helper below the type map:
# Maps enum class names to their source lookup tables
ENUM_TABLES = {
    "ExperimentStatus": "lu_statuses",
    "ExperimentMethod": "lu_methods",
    "AssignmentUnit": "lu_assignment_units",
    "DecisionOutcome": "lu_decision_outcomes",
    "MetricType": "lu_metric_types",
    "ReadoutType": "lu_readout_types",
}


def execute_sql(w, sql):
    """Run a SQL statement against the warehouse and return rows as dicts."""
    response = w.statement_execution.execute_statement(
        warehouse_id=WAREHOUSE_ID, statement=sql, wait_timeout="50s"
    )
    if response.status.state.value == "FAILED":
        raise RuntimeError(f"SQL failed: {response.status.error.message}")
    columns = [col.name for col in response.manifest.schema.columns]
    rows = []
    if response.result and response.result.data_array:
        for row_data in response.result.data_array:
            rows.append(dict(zip(columns, row_data)))
    return rows

What does this code do?

ENUM_TABLES maps each generated enum class name back to its source lu_* table. The validator uses this to check enum drift.

execute_sql() is the same Statement Execution API pattern from your generator. It runs a query via your SQL warehouse and returns the results as dictionaries.

  • Save schema_validator.py.
  • Confirm the module loads by running:
python -c "import schema_validator; print('Module loaded OK')"

You should see Module loaded OK with no import errors.

Seeing a ModuleNotFoundError?

If the error mentions schemas, confirm that your generated schemas.py is in the same directory as schema_validator.py.

If it mentions databricks, ensure your DATABRICKS_WAREHOUSE_ID environment variable is set in this terminal session.

Help me debug my schema_validator.py import error

  • Now add the helper functions that extract and compare types. Paste this below execute_sql:
def get_model_fields(model):
    """Extract field names and their base Python type strings."""
    fields = {}
    for name, field_info in model.model_fields.items():
        fields[name] = unwrap_optional(field_info.annotation)
    return fields


def unwrap_optional(annotation):
    """Get the base type name, stripping Optional wrapper if present."""
    origin = getattr(annotation, "__origin__", None)
    if origin is type(None):
        return "None"
    args = getattr(annotation, "__args__", None)
    if args and type(None) in args:
        inner = [a for a in args if a is not type(None)]
        if inner:
            return inner[0].__name__ if hasattr(inner[0], "__name__") else str(inner[0])
    if hasattr(annotation, "__name__"):
        return annotation.__name__
    return str(annotation)


def type_compatible(python_type, sql_type):
    """Check if a Python type is compatible with a SQL type."""
    base_sql = re.split(r"[<(]", sql_type.upper())[0].strip()
    for py_type, sql_types in SQL_TYPE_MAP.items():
        if python_type == py_type and base_sql in sql_types:
            return True
    # Enum subclasses and unknown types map to STRING
    if python_type not in SQL_TYPE_MAP and base_sql == "STRING":
        return True
    return False


def class_name_to_table_name(class_name):
    """Convert PascalCase back to snake_case for table lookup."""
    s = re.sub(r"([A-Z])", r"_\1", class_name).lower().lstrip("_")
    return s

What do these helpers do?

  • get_model_fields() reads each Pydantic model's fields and extracts the base Python type name.
  • unwrap_optional() strips the Optional[...] wrapper so you compare the inner type.
  • type_compatible() checks whether a Python type matches a SQL type using the reverse map. Enum subclasses default to STRING compatibility.
  • class_name_to_table_name() reverses the PascalCase conversion so Experiment becomes experiment.
  • Add the core validation function and model discovery below the helpers:
def validate_table(w, model, table_name):
    """Two-directional check: model vs table, table vs model."""
    errors = []
    sql = f"""
    SELECT column_name, data_type, is_nullable
    FROM {CATALOG}.information_schema.columns
    WHERE table_catalog = '{CATALOG}'
      AND table_schema = '{SCHEMA}'
      AND table_name = '{table_name}'
    ORDER BY ordinal_position
    """
    live_columns = execute_sql(w, sql)
    live_map = {row["column_name"]: row for row in live_columns}
    model_fields = get_model_fields(model)
    # Forward check: every model field must exist in table with compatible type
    for field_name, py_type in model_fields.items():
        if field_name not in live_map:
            errors.append(f"  FORWARD: field '{field_name}' in model but not in table")
        else:
            sql_type = live_map[field_name]["data_type"]
            if not type_compatible(py_type, sql_type):
                errors.append(f"  FORWARD: field '{field_name}' type mismatch: model={py_type}, table={sql_type}")
    # Reverse check: every non-nullable column must exist in model
    for col_name, col_info in live_map.items():
        if col_info["is_nullable"] == "NO" and col_name not in model_fields:
            errors.append(f"  REVERSE: non-nullable column '{col_name}' in table but not in model")
    return errors


def discover_models():
    """Auto-discover all BaseModel subclasses in the schemas module."""
    from pydantic import BaseModel
    models = []
    for name, obj in inspect.getmembers(schemas, inspect.isclass):
        if issubclass(obj, BaseModel) and obj is not BaseModel:
            models.append((name, obj))
    return models

Why two directions?

The forward check catches stale models. If someone renames a column in the Delta table, the model still has the old name and the validator flags it.

The reverse check catches untracked additions. If someone adds a new non-nullable column to a table, the model does not know about it and the validator flags it. Together, these two directions ensure nothing slips through.

  • Save schema_validator.py.
  • Confirm the validator discovers all 17 models by running:
python -c "from schema_validator import discover_models; print(len(discover_models()), 'models discovered')"

You should see 17 models discovered.

Seeing fewer than 17 models?

If you see 0, confirm that schemas.py exists in the same directory and was generated by running schemas_generator.py in the previous step.

If you see a number other than 17, re-run python schemas_generator.py to regenerate from the latest table metadata.

Help me figure out why discover_models is not returning 17 models

Add enum validation to the drift gate

Table structure is only half the contract. If a new value gets added to a lookup table but the enum is not regenerated, downstream code that switches on enum members will silently miss the new case.

  • Add the validate_enums function below discover_models in schema_validator.py:
def validate_enums(w):
    """Check that generated enums match live lu_* values (full-set equality)."""
    errors = []
    for enum_class_name, table_name in ENUM_TABLES.items():
        enum_class = getattr(schemas, enum_class_name, None)
        if enum_class is None:
            errors.append(f"  ENUM: class '{enum_class_name}' not found in schemas")
            continue
        sql = f"SELECT DISTINCT value FROM {CATALOG}.{SCHEMA}.{table_name} ORDER BY value"
        rows = execute_sql(w, sql)
        live_values = {row["value"] for row in rows if row["value"]}
        enum_values = {member.value for member in enum_class}
        # Bidirectional set comparison
        missing_in_enum = live_values - enum_values
        extra_in_enum = enum_values - live_values
        if missing_in_enum:
            errors.append(f"  ENUM: {enum_class_name} missing values from {table_name}: {missing_in_enum}")
        if extra_in_enum:
            errors.append(f"  ENUM: {enum_class_name} has extra values not in {table_name}: {extra_in_enum}")
    return errors

What does this code do?

For each of the six enum classes, validate_enums queries the live lookup table and compares the full value set against the generated enum members. It checks both directions: values present in the table but missing from the enum, and values in the enum that no longer exist in the table.

This catches the scenario where someone adds a new status or method to a lookup table but forgets to regenerate schemas.py.

  • Save schema_validator.py.
  • Verify enum validation loads cleanly by running:
python -c "from schema_validator import validate_enums; print('validate_enums ready')"

You should see validate_enums ready with no errors.

Wire into CI and demonstrate failure injection

The individual check functions exist, but without an orchestrator that exits with a non-zero code, CI cannot gate on the result. The main() function ties everything together and speaks the language CI understands: exit 0 for pass, exit 1 for fail.

  • Add the main function and entry point at the bottom of schema_validator.py:
def main():
    w = WorkspaceClient()
    models = discover_models()
    all_passed = True
    print(f"Validating {len(models)} models against live schema...\n")
    # Validate each model against its live table
    for class_name, model in sorted(models):
        table_name = class_name_to_table_name(class_name)
        errors = validate_table(w, model, table_name)
        if errors:
            print(f"FAIL: {class_name} ({table_name})")
            for err in errors:
                print(err)
            all_passed = False
        else:
            print(f"PASS: {class_name} ({table_name})")
    # Validate enums against live lookup tables
    print("\nValidating enums against live lookup tables...")
    enum_errors = validate_enums(w)
    if enum_errors:
        print("FAIL: Enum validation")
        for err in enum_errors:
            print(err)
        all_passed = False
    else:
        print("PASS: Enum validation")
    print()
    if all_passed:
        print("ALL CHECKS PASSED")
        sys.exit(0)
    else:
        print("VALIDATION FAILED")
        sys.exit(1)


if __name__ == "__main__":
    main()

What does this code do?

  • discover_models() finds all 17 Pydantic models in schemas.py automatically.
  • For each model, it converts the class name to the snake_case table name and runs two-directional validation.
  • After table checks, it validates all six enums against live lookup tables.
  • sys.exit(0) on success and sys.exit(1) on failure is the standard contract CI pipelines expect.
  • Save schema_validator.py.

Before you run this, what do you expect to see? Your models were generated from the same tables the validator will check against. Think about whether you expect all 17 to pass or if any might fail.

  • Run the full validator:
python schema_validator.py

You should see 17 PASS lines (one per model), followed by PASS: Enum validation, and finally ALL CHECKS PASSED.

Seeing FAIL on a model?

If you see a FORWARD error, the model has a field the live table does not. This means schemas.py is stale. Regenerate it by running python schemas_generator.py then re-run the validator.

If you see a REVERSE error, the table has a non-nullable column missing from the model. Same fix: regenerate.

Help me debug my schema_validator.py failures

Now prove the gate actually catches drift. You will inject a fake field into one of your models, re-run the validator, and watch it fail.

  • Open schemas.py and find any model class (for example, the first one alphabetically).
  • Add a line inside the class body:
    fake_field: str
  • Save schemas.py.

What do you think will happen when you run the validator now? The model claims a field called fake_field exists, but the live table has no such column.

  • Run the validator again:
python schema_validator.py; echo "Exit code: $?"

You should see a FAIL line for the model you edited, with FORWARD: field 'fake_field' in model but not in table, followed by VALIDATION FAILED and exit code 1.

Why does this matter for CI?

Exit code 1 is the universal signal for failure in CI systems. Any CI runner will mark the build as failed and block the merge. This single exit code is what turns your schema contract from documentation into enforcement.

  • Remove the fake_field: str line from schemas.py.
  • Save schemas.py and re-run the validator to confirm a clean pass:
python schema_validator.py; echo "Exit code: $?"

You should see all 17 PASS lines, ALL CHECKS PASSED, and exit code 0.

  • Wire schema_validator.py into your existing CI pipeline alongside the stats test suite. Add it as a step that runs after your tests:
python schema_validator.py

Where does this go in CI?

Add this command to your CI configuration (Makefile, GitHub Actions YAML, or Databricks workflow) as a step that runs after your existing stats tests. Because it exits 1 on any mismatch, the CI run will fail and block the merge.

You now have two automated gates: your stats test suite validates computational correctness, and schema_validator.py validates structural correctness.

✔️ Awesome, I've got everything!

Great. Double check you have saved schema_validator.py and that your last run showed ALL CHECKS PASSED with exit code 0.

ⓧ I'd like to double check the full code

Here is the complete schema_validator.py. Scroll through and confirm each section is present in order: imports, constants, SQL_TYPE_MAP, ENUM_TABLES, execute_sql, get_model_fields, unwrap_optional, type_compatible, class_name_to_table_name, validate_table, validate_enums, discover_models, main.

"""
MES Schema Validator (CI Drift Gate)
Two-directional schema validation + enum drift detection.
Exit 0 = all pass, Exit 1 = any failure.
"""
import inspect
import os
import re
import sys
from enum import Enum

from databricks.sdk import WorkspaceClient

import schemas

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"
WAREHOUSE_ID = os.environ["DATABRICKS_WAREHOUSE_ID"]
SQL_TYPE_MAP = {
    "str": ["STRING"],
    "int": ["INT", "BIGINT", "INTEGER", "SMALLINT", "TINYINT"],
    "float": ["DOUBLE", "FLOAT", "DECIMAL"],
    "bool": ["BOOLEAN"],
    "date": ["DATE"],
    "datetime": ["TIMESTAMP"],
    "list": ["ARRAY"],
    "dict": ["MAP", "STRUCT"],
}

ENUM_TABLES = {
    "ExperimentStatus": "lu_statuses",
    "ExperimentMethod": "lu_methods",
    "AssignmentUnit": "lu_assignment_units",
    "DecisionOutcome": "lu_decision_outcomes",
    "MetricType": "lu_metric_types",
    "ReadoutType": "lu_readout_types",
}
def execute_sql(w, sql):
    response = w.statement_execution.execute_statement(
        warehouse_id=WAREHOUSE_ID, statement=sql, wait_timeout="50s"
    )
    if response.status.state.value == "FAILED":
        raise RuntimeError(f"SQL failed: {response.status.error.message}")
    columns = [col.name for col in response.manifest.schema.columns]
    rows = []
    if response.result and response.result.data_array:
        for row_data in response.result.data_array:
            rows.append(dict(zip(columns, row_data)))
    return rows
def get_model_fields(model):
    fields = {}
    for name, field_info in model.model_fields.items():
        fields[name] = unwrap_optional(field_info.annotation)
    return fields


def unwrap_optional(annotation):
    origin = getattr(annotation, "__origin__", None)
    if origin is type(None):
        return "None"
    args = getattr(annotation, "__args__", None)
    if args and type(None) in args:
        inner = [a for a in args if a is not type(None)]
        if inner:
            return inner[0].__name__ if hasattr(inner[0], "__name__") else str(inner[0])
    if hasattr(annotation, "__name__"):
        return annotation.__name__
    return str(annotation)


def type_compatible(python_type, sql_type):
    base_sql = re.split(r"[<(]", sql_type.upper())[0].strip()
    for py_type, sql_types in SQL_TYPE_MAP.items():
        if python_type == py_type and base_sql in sql_types:
            return True
    if python_type not in SQL_TYPE_MAP and base_sql == "STRING":
        return True
    return False


def class_name_to_table_name(class_name):
    s = re.sub(r"([A-Z])", r"_\1", class_name).lower().lstrip("_")
    return s
def validate_table(w, model, table_name):
    errors = []
    sql = f"""
    SELECT column_name, data_type, is_nullable
    FROM {CATALOG}.information_schema.columns
    WHERE table_catalog = '{CATALOG}'
      AND table_schema = '{SCHEMA}'
      AND table_name = '{table_name}'
    ORDER BY ordinal_position
    """
    live_columns = execute_sql(w, sql)
    live_map = {row["column_name"]: row for row in live_columns}
    model_fields = get_model_fields(model)
    for field_name, py_type in model_fields.items():
        if field_name not in live_map:
            errors.append(f"  FORWARD: field '{field_name}' in model but not in table")
        else:
            sql_type = live_map[field_name]["data_type"]
            if not type_compatible(py_type, sql_type):
                errors.append(f"  FORWARD: field '{field_name}' type mismatch: model={py_type}, table={sql_type}")
    for col_name, col_info in live_map.items():
        if col_info["is_nullable"] == "NO" and col_name not in model_fields:
            errors.append(f"  REVERSE: non-nullable column '{col_name}' in table but not in model")
    return errors
def validate_enums(w):
    errors = []
    for enum_class_name, table_name in ENUM_TABLES.items():
        enum_class = getattr(schemas, enum_class_name, None)
        if enum_class is None:
            errors.append(f"  ENUM: class '{enum_class_name}' not found in schemas")
            continue
        sql = f"SELECT DISTINCT value FROM {CATALOG}.{SCHEMA}.{table_name} ORDER BY value"
        rows = execute_sql(w, sql)
        live_values = {row["value"] for row in rows if row["value"]}
        enum_values = {member.value for member in enum_class}
        missing_in_enum = live_values - enum_values
        extra_in_enum = enum_values - live_values
        if missing_in_enum:
            errors.append(f"  ENUM: {enum_class_name} missing values from {table_name}: {missing_in_enum}")
        if extra_in_enum:
            errors.append(f"  ENUM: {enum_class_name} has extra values not in {table_name}: {extra_in_enum}")
    return errors


def discover_models():
    from pydantic import BaseModel
    models = []
    for name, obj in inspect.getmembers(schemas, inspect.isclass):
        if issubclass(obj, BaseModel) and obj is not BaseModel:
            models.append((name, obj))
    return models
def main():
    w = WorkspaceClient()
    models = discover_models()
    all_passed = True
    print(f"Validating {len(models)} models against live schema...\n")
    for class_name, model in sorted(models):
        table_name = class_name_to_table_name(class_name)
        errors = validate_table(w, model, table_name)
        if errors:
            print(f"FAIL: {class_name} ({table_name})")
            for err in errors:
                print(err)
            all_passed = False
        else:
            print(f"PASS: {class_name} ({table_name})")
    print("\nValidating enums against live lookup tables...")
    enum_errors = validate_enums(w)
    if enum_errors:
        print("FAIL: Enum validation")
        for err in enum_errors:
            print(err)
        all_passed = False
    else:
        print("PASS: Enum validation")
    print()
    if all_passed:
        print("ALL CHECKS PASSED")
        sys.exit(0)
    else:
        print("VALIDATION FAILED")
        sys.exit(1)


if __name__ == "__main__":
    main()

Your MES data layer now has an automated guardian. Any schema drift, field removal, type mismatch, or enum addition that is not reflected in the contract will block CI before it reaches production. Next up, you will extend this into a schema evolution report that tells you exactly what changed.

Secret mission

Schema Evolution Report

Your drift gate exits 0 or 1, but it never tells you what actually changed. Extend the validator to produce a markdown diff report showing exactly what drifted since the last clean run. Column additions, removals, type changes, and enum value drift all appear in a single readable report.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely against your existing SQL warehouse and local files, so there are no ongoing costs.

Resources you used:

  • Local files: schemas_generator.py, schema_validator.py, schemas.py, schema_snapshot.json, schema_evolution_report.md.
  • Installed package: pydantic==2.13.4.

Keep everything running

No action needed. Choose this if you are still actively building on the MES schema contracts or plan to extend the validator further.

  • Your schemas_generator.py and schema_validator.py remain ready to run in CI.
  • Your schemas.py continues to enforce schema contracts on every pipeline run.
  • The pydantic package stays installed in your Python environment.

Pause - I'll come back to this later

There are no running processes to stop. Your scripts only execute when called directly or triggered by CI.

  • If you added schema_validator.py to your CI pipeline config, you can comment out or remove that step to prevent it from running on every commit.
  • All local files remain on disk and can be re-enabled at any time.

Delete - I don't want to use this again

Remove all project files and the installed package to start fresh.

  • Delete the local project files by running this command (macOS/Linux):
rm schemas_generator.py schema_validator.py schemas.py schema_snapshot.json schema_evolution_report.md
  • On Windows, run this instead:
Remove-Item schemas_generator.py, schema_validator.py, schemas.py, schema_snapshot.json, schema_evolution_report.md
  • Uninstall the Pydantic package by running:
pip uninstall pydantic -y
  • If you added schema_validator.py to your CI pipeline config, remove that step from the config file.
  • Your existing Databricks workspace, SQL warehouse, and MES tables are unchanged. They were pre-existing resources and remain as they were before this project.

Nice Work!

Nice Work!

You just built a production-grade schema contract layer for your Marketing Experimentation System. Your 17 Delta tables are now protected by auto-generated Pydantic models, introspected enums, and a CI drift gate that catches regressions before they reach production.

Here's what you accomplished:

  • Auto-generated schemas.py with Pydantic models for all 17 MES tables, with live enums introspected from lu_* lookup tables and B2B field validators enforcing experiment design constraints.
  • Built a two-directional CI drift gate in schema_validator.py that validates model-to-table, table-to-model, and enum drift, exiting 1 on any mismatch to fail CI.
  • Proved the gate catches regressions with a fake-field injection test, wired both scripts into your existing CI pipeline alongside the stats test suite.
  • Secret Mission: Extended the validator with a schema evolution report that produces a markdown diff of added columns, removed columns, type changes, and enum value drift since the last clean run.

Ready to quiz yourself?