🛡️ Data Contract Validator

Your dbt models don't stop at the warehouse. This checks that whatever they feed — a CRM, an API — still gets the shape it expects, and fails the PR when it doesn't.

🎯 The problem

dbt gives you tests, git history and code review for everything up to the warehouse. Past that, nothing.

Your models get synced into a CRM. They get read by an API. Those destinations have schemas too, and they drift:

• Someone renames a property in a HubSpot settings page. No pull request, no git history, no review — a dropdown and a save button. Your sync keeps running and quietly writes into a field that no longer means what it used to. Nobody finds out until a campaign goes out against bad data.
• Your analytics team drops a column. The FastAPI service still expects it. Nobody notices until production 500s at 2 AM.

Neither shows up in your dbt tests, because dbt doesn't know those destinations exist. The contract between them lives in someone's memory.

This tool sits on that boundary. It reads what your dbt models actually produce, reads what the destination actually expects, compares them, and fails CI when they've drifted apart.

   dbt models                Data Contract Validator              destinations
(what the pipeline  ──▶  extract → normalize → compare  ◀──  HubSpot CRM
    produces)                      ↓                          FastAPI / Pydantic
                        critical issues block the build

Built for trust

A check that gates a deploy is only useful if it doesn't cry wolf:

• Canonical types — dbt varchar and Pydantic str are understood to be the same thing, so you don't get drowned in fake "type mismatch" warnings.
• A real SQL parser (sqlglot) instead of regex — CTEs, || concatenation, window functions and quoted identifiers are parsed correctly.
• Confidence-aware — if the tool can't fully resolve a model's columns (e.g. SELECT *), it will warn rather than falsely block your build.

⚡ Quick start

pip install data-contract-validator

# Initialize config + CI workflow in your dbt project
contract-validator init --interactive

# Sanity-check the setup
contract-validator test

# Validate
contract-validator validate

Guarding a reverse-ETL sync into HubSpot takes one block of config:

target:
  hubspot:
    type: "hubspot"
    object_type: "contacts"     # or companies, deals, or a custom object
    fields:                     # the properties your sync actually writes
      - email
      - lifecyclestage
      - lifetime_value

export HUBSPOT_ACCESS_TOKEN=pat-xxxx    # HubSpot Private App token
contract-validator validate

🚀 Getting started, step by step

If you're setting this up on a project for the first time, the order below avoids the sharp edges:

1. Install into the same environment dbt runs in (not a separate venv) — the tool needs to see your dbt project:

   pip install data-contract-validator

   Already have .retl-validator.yml committed by a teammate? Skip to step 5.

2. Generate the config + CI workflow (one-time):

   contract-validator init --interactive

   You'll be asked: where your dbt project is, which destination you're validating against, whether your models live in this local project or a different GitHub repo, and then the local path (or the org/repo + path within it, plus an optional branch/tag/commit — blank reads the repo's default branch). Local-vs-GitHub is asked explicitly rather than guessed from the path's shape — a local path like app/models is syntactically identical to a GitHub org/repo string, so there's no reliable way to infer which one you mean. If you pick GitHub, it checks the path actually exists before writing the config — so a typo surfaces here instead of at validate time.

   init refuses to touch an existing .retl-validator.yml or workflow file — it won't clobber hand-added mapping entries just because you upgraded the package and re-ran init. Pass --force if you really want to regenerate them from the new version's defaults.

3. Pre-commit hook: init --interactive asks whether you want one set up right after creating the config and CI workflow — say yes there and it's done. To add one later (or if you used non-interactive init, which doesn't prompt), run it standalone:

   contract-validator setup-precommit --install-hooks

4. If the target repo is private, set a token before running anything that talks to GitHub locally:

   export GITHUB_TOKEN=$(gh auth token)   # or a PAT with repo read access

   See Private GitHub repos need GITHUB_TOKEN below for why this is easy to miss.

5. Sanity-check the setup:

   contract-validator test

   Confirms the config parses, the dbt project is found, and the target (local path, GitHub path, or CRM connection) is reachable. If this fails, validate will fail the same way — fix it here first.

6. Run it:

   contract-validator validate

7. When it reports a critical issue, diagnose before assuming your dbt model is wrong:

   • Real missing column/table → fix the dbt model.
   • Target name doesn't match the dbt model by convention (renamed/prefixed) → add an entry under mapping.tables in .retl-validator.yml (see When do I need mapping?).
   • A table that's genuinely populated by something other than dbt (e.g. a separate streaming pipeline) and has no source model on purpose → add it to mapping.exclude. table=True alone is not used to infer this automatically — see FastAPI / Pydantic for why.

8. For accurate type-checking (not just column-presence checks), run dbt docs generate before validate so it picks up catalog.json (Tier 1, real warehouse types) instead of inferring from SQL text — see How extraction works below.

One-off validation (no config file)

# Local dbt project against a local Pydantic models file or directory
contract-validator validate \
  --dbt-project ./my-dbt-project \
  --fastapi-local ./my-api/app/models.py

# dbt project against models in another GitHub repo (microservices)
contract-validator validate \
  --dbt-project . \
  --fastapi-repo "my-org/my-api" \
  --fastapi-path "app/models.py"

# ...against a dev/staging branch of that repo instead of its default branch
contract-validator validate \
  --dbt-project . \
  --fastapi-repo "my-org/my-api" \
  --fastapi-path "app/models.py" \
  --fastapi-ref "dev"

--fastapi-ref accepts a branch, tag, or commit SHA. It's useful for validating an in-progress API change (on a dev or feature branch) against dbt before it merges to main — catch the break in the PR that's about to introduce it, not after.

Branch auto-matching

You usually don't need to set a ref at all. When none is given, the branch is matched automatically so a dbt change headed for dev checks against the API repo's dev branch — keeping each environment validated against its own counterpart with no per-branch config:

Where it's running	Branch used
A GitHub Actions pull request	the branch the PR targets (GITHUB_BASE_REF) — the environment the change is heading toward, not the feature branch it's coming from
A GitHub Actions push	the pushed branch (GITHUB_REF_NAME)
Locally	your dbt project's current git branch

If that branch doesn't exist on the target repo — common when the API repo doesn't mirror your dbt repo's branch names — it silently falls back to the target's default branch, so this never turns into a spurious failure. An explicit --fastapi-ref or target.*.ref always overrides auto-matching.

🎯 Destinations

HubSpot CRM (reverse ETL)

A CRM is the far end of reverse ETL, and the least guarded surface in the whole pipeline. Unlike a codebase, its schema lives in an admin UI, editable by anyone with the right permissions, with no code review and no git history. A property renamed in a settings page breaks a sync exactly the way a dropped dbt column does — and nothing in your repo would show it.

target:
  hubspot:
    type: "hubspot"
    object_type: "contacts"     # or companies, deals, or a custom object
    fields:                     # the properties your sync actually writes
      - email
      - lifecyclestage
      - lifetime_value

export HUBSPOT_ACCESS_TOKEN=pat-xxxx    # HubSpot Private App token
contract-validator validate

Two things differ from a code target, both deliberate:

• fields is strongly recommended. A stock HubSpot object carries 100–400+ properties, nearly all irrelevant to any one sync. Comparing against all of them reintroduces exactly the noise the canonical type system exists to eliminate. Omit fields and every writable property is used — fine for exploring, not for a check that gates a deploy.
• Calculated, hidden, and read-only properties are always excluded. A sync could never populate them, so flagging them would be a permanent, unfixable failure.

Because HubSpot properties have no schema-level "required" flag, a missing field is a warning by default. Use mapping.critical_columns to mark the ones that are genuinely load-bearing for your sync so they fail the build instead.

🔑 The token is read from HUBSPOT_ACCESS_TOKEN, never stored in .retl-validator.yml — that file is meant to be committed. Create one under HubSpot Settings → Integrations → Private Apps with the crm.schemas.<object>.read scope.

FastAPI / Pydantic

Pydantic / SQLModel classes are parsed from source with Python's ast (no imports executed). Optional[...] controls whether a field is required. An explicit __tablename__ is used as the table name when present; otherwise the class name is converted to snake_case.

table=True SQLModel classes are validated the same as any other class — they are not skipped. Whether a table is meant to come from dbt is business knowledge that isn't recoverable from the Python source: two structurally identical table=True classes can need opposite treatment (one is a normal dbt-fed table your API also returns directly; another is populated by a Kafka stream and was never meant to have a dbt model). Use mapping.exclude to state the latter case explicitly rather than relying on table=True to imply it.

🔍 How extraction works (and why it's accurate)

dbt side — tiered, best-source-wins

Tier	Source	Types	Confidence	Notes
1	target/catalog.json	Real warehouse types	high	Produced by dbt docs generate. Most accurate.
2	sqlglot SQL parse	Inferred (often unknown)	medium	Trusted column names; enriched with documented types from manifest.json. Detects SELECT *.
3	regex parse	Guessed	low	Last resort. Never used to hard-fail a build.

The tool auto-detects what's available and degrades gracefully — so it works offline in pre-commit and with full type fidelity in a warehouse-connected CI job.

💡 Tip: run dbt docs generate in CI before validating to unlock Tier 1 (real types). Without it, you still get accurate column-presence checks from Tier 2. The workflow init generates includes this step already, commented out — it needs your warehouse adapter and credentials filled in, which can't be guessed, so it isn't active by default.

🚦 What gets flagged

Severity	Meaning	Example
🚨 Critical	Blocks the build	Destination requires a column the dbt model no longer produces
⚠️ Warning	Worth a look, non-blocking	A real type mismatch, or a missing column on a model we couldn't fully resolve

$ contract-validator validate

🛡️ Data Contract Validation Results:
Status: ❌ FAILED
Critical: 1 | Warnings: 0

🚨 Critical Issues (Must Fix):
  💥 user_analytics
     Column: total_orders
     Problem: Target REQUIRES column 'total_orders' but source doesn't provide it
     🔧 Fix: Add column 'total_orders' to source model for table 'user_analytics'

🔧 Configuration (.retl-validator.yml)

version: "1.0"
name: "my-project-contracts"

source:
  dbt:
    project_path: "."
    auto_compile: true
    # Force Tier 2/3 SQL parsing even if catalog/manifest exist:
    disable_manifest: false

target:
  # A reverse-ETL destination...
  hubspot:
    type: "hubspot"
    object_type: "contacts"
    fields: [email, lifecyclestage, lifetime_value]

  # ...and/or a code target.
  fastapi:
    # GitHub repo:
    type: "github"
    repo: "my-org/my-api"
    path: "app/models.py"
    # Optional: pin a branch, tag, or commit to read from. Omit this and the
    # branch is auto-matched (PR target branch in CI, current branch locally),
    # falling back to the repo's default branch -- see "Branch auto-matching".
    # Set it only to force one specific ref regardless of context.
    # ref: "dev"
    # ...or local:
    # type: "local"
    # path: "../my-api/app/models.py"

# Optional: explicit mapping for when names don't line up by convention.
mapping:
  tables:
    # target table : source (dbt) model
    user_analytics: user_analytics_summary
  columns:
    user_analytics:
      # target column : source column
      userId: user_id
  # Target tables with no source model on purpose (e.g. Kafka-populated,
  # not dbt) -- see "When do I need mapping?" below.
  exclude:
    - feed_interaction
  # Missing columns that should fail the build rather than warn.
  critical_columns:
    contacts:
      - email

validation:
  fail_on: ["missing_tables", "missing_required_columns"]
  warn_on: ["type_mismatches", "missing_optional_columns"]

Private GitHub repos need GITHUB_TOKEN

If target.*.repo points at a private repository, contract-validator needs a token with read access to it. Where that token comes from is different locally vs. in CI — and the CI case has a sharp edge worth understanding before it silently fails on a PR.

Locally, set the GITHUB_TOKEN environment variable before running the CLI. On bash/zsh that's export (there's nothing to install — export just makes the variable visible to the contract-validator process you run next):

export GITHUB_TOKEN=$(gh auth token)   # or a PAT with repo read access
contract-validator validate

GitHub's API 404s (not 403s) an unauthenticated request to a private path, so without a token this looks identical to a plain typo in path — contract-validator init --interactive and contract-validator test both check target.*.path actually exists and will point you at this if the lookup 404s with no token set.

In CI, the workflow init generates for a GitHub target wires up GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} — the token Actions auto-provides. That's fine and needs no setup if your target repo is public. But that token only has access to the repository the workflow is running in, so if your dbt repo and your API repo are different repos and the target is private, it silently can't read it, and validation fails on every PR with no clue why.

⚠️ Strong recommendation: if your target repo is private, switch this to a token you create yourself before you rely on this workflow. The generated workflow carries a loud comment for exactly this — don't wait to discover it the hard way on a PR.

To make that switch:

1. Create a token with read access to the target repo — a fine-grained PAT scoped to just that repo's Contents (read-only) is the least-privilege option; a classic PAT with the repo scope also works.

2. In the repo running the workflow (your dbt repo): Settings → Secrets and variables → Actions → New repository secret. Name it API_REPO_TOKEN (or similar) and paste the token as the value.

   ⚠️ GitHub rejects any secret name starting with GITHUB_ — it's a reserved prefix. You cannot create a secret literally called GITHUB_TOKEN; that's not a naming suggestion, the UI will refuse it. That's why the secret needs a different name, even though the environment variable it feeds is GITHUB_TOKEN — two different things with confusingly similar names:

   env:
     GITHUB_TOKEN: ${{ secrets.API_REPO_TOKEN }}
   #  ^^^^^^^^^^^   local variable name, can be anything -- the CLI
   #                just needs it called GITHUB_TOKEN to find it
   #                            ^^^^^^^^^^^^^^ the *secret's* name --
   #                            this is what GitHub restricts

3. Replace secrets.GITHUB_TOKEN with secrets.API_REPO_TOKEN (or whatever you named it) in the workflow's env: block.

Skip all of this for a local or CRM target — init omits the whole env: block, since neither talks to the GitHub API at all.

When do I need mapping?

Most of the time you don't. Names are matched automatically across:

• snake_case / camelCase / casing — UserAnalytics → user_analytics, userId → user_id
• plural ↔ singular — dbt's plural users matches Pydantic's User (→ user) with no config (and it won't over-match — address is never confused with addres).

Reach for mapping.tables / mapping.columns only when a model or column is named so differently that convention can't bridge it (e.g. Pydantic user_id ↔ dbt customer_identifier).

mapping.exclude is different — it's not about renamed models, it's for a target table that has no source model on purpose, because it's populated by something other than dbt (a Kafka stream, a cron job, etc.). This can't be inferred from the code (a table=True SQLModel class looks identical whether or not dbt is supposed to feed it), so it has to be a deliberate, human-stated exception:

mapping:
  exclude:
    - feed_interaction
    - affiliate_reward

Anything not listed is validated normally — including table=True classes, which are treated the same as any other target and are not silently skipped.

mapping.critical_columns is the mirror image of exclude: it escalates a missing column to build-failing CRITICAL even when the extractor reported it as optional. Some targets have no schema-level "required" concept at all — a HubSpot property has no required flag — so without this, every missing field there could only ever warn:

mapping:
  critical_columns:
    contacts:
      - email            # the sync is meaningless without these
      - lifecyclestage

Like exclude, this is business knowledge no extractor can infer, and it works for any target type.

🐍 Python API

from data_contract_validator import ContractValidator, DBTExtractor, FastAPIExtractor

dbt = DBTExtractor(project_path="./dbt-project")
fastapi = FastAPIExtractor.from_github_repo("my-org/my-api", "app/models.py")

validator = ContractValidator(
    source_extractor=dbt,
    target_extractor=fastapi,
    mapping={"tables": {"user_analytics": "user_analytics_summary"}},  # optional
)
result = validator.validate()

if not result.success:
    for issue in result.critical_issues:
        print(f"💥 {issue.table}.{issue.column}: {issue.message}")

🪝 CI / pre-commit integration

GitHub Actions

contract-validator init generates a workflow for you. Minimal version:

name: 🛡️ Data Contract Validation
on:
  pull_request:
    paths: ["models/**/*.sql", "dbt_project.yml", "**/*models*.py"]
jobs:
  validate-contracts:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v4
        with: { python-version: "3.11" }
      - run: pip install data-contract-validator
      # Optional: `dbt docs generate` here for real warehouse types (Tier 1)
      - run: contract-validator validate --output github
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          HUBSPOT_ACCESS_TOKEN: ${{ secrets.HUBSPOT_ACCESS_TOKEN }}

GITHUB_TOKEN here is only needed if target is a github repo (init omits the whole env: block for a local target). The default above works as-is for a public target repo. For a private one, strongly recommended: swap it for a token you create yourself — see Private GitHub repos need GITHUB_TOKEN above for why the default silently can't read a private target, and how to set up the replacement.

Pre-commit

contract-validator setup-precommit --install-hooks

repos:
  - repo: https://github.com/OGsiji/data-contract-validator
    rev: v1.3.0
    hooks:
      - id: contract-validation

🧪 Output formats

contract-validator validate --output terminal   # human-friendly (default)
contract-validator validate --output json        # machine-readable for CI
contract-validator validate --output github       # GitHub Actions annotations

🚀 Supported sources and destinations

Source: dbt (all adapters — Snowflake, BigQuery, Redshift, Postgres, …). Destinations: HubSpot CRM, FastAPI (Pydantic v2 + SQLModel).

The extractor architecture is intentionally pluggable (BaseExtractor → Dict[str, Schema] with canonical types), so additional sources and destinations can be added without touching the validator. Open an issue to request one.

🛠️ Development & testing

git clone https://github.com/OGsiji/data-contract-validator
cd data-contract-validator

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"     # or: pip install -e ".[test]"

# Run the suite
pytest

# Lint / format
black data_contract_validator tests

The test suite covers the canonical type system (tests/test_core/test_types.py), the tiered dbt extractor including sqlglot CTE handling and catalog.json (tests/test_extractors/test_dbt.py), and the confidence/mapping behavior of the validator (tests/test_core/test_validator.py).

Adding an extractor

from data_contract_validator.extractors.base import BaseExtractor
from data_contract_validator.core.types import CanonicalType

class MyExtractor(BaseExtractor):
    def extract_schemas(self):
        # return Dict[str, Schema]; use self._make_column(...) so each column
        # carries a canonical_type the validator can compare.
        ...

🗺️ Roadmap

• More reverse-ETL destinations (Salesforce, Braze, Customer.io)
• Real compatibility semantics (nullability, additive vs. breaking changes)
• Reporter/logging abstraction (quiet/embeddable core)
• A canonical, language-neutral contract artifact + baseline/snapshot diffing
• More code targets (Django, SQLAlchemy, GraphQL, OpenAPI)

📄 License

MIT — see LICENSE.

🆘 Support

• 🐛 Issues: https://github.com/OGsiji/data-contract-validator/issues
• 📧 Email: ogunniransiji@gmail.com

If this saves you a production incident, please ⭐ the repo.