datacontract-framework 1.0.0

datacontract-framework — Lightweight Data Contract Framework for automated data quality enforcement in lakehouse pipelines

DCF — DataContract Framework

A lightweight, self-hostable framework for defining and automatically enforcing data quality contracts on batch datasets. Built for data engineering pipelines — works standalone, in Airflow, or any Python environment.

---

Table of Contents

1. What DCF Does
2. Installation
3. Quick Start (5 minutes)
4. Full Workflow
   • Step 1 — Write a Contract
   • Step 2 — Check the Contract
   • Step 3 — Start the Registry and Dashboard
   • Step 4 — Publish the Contract
   • Step 5 — Validate in Your Pipeline
   • Step 6 — Use in Airflow
   • Step 7 — Observe Breaches
5. Contract YAML Reference
6. CLI Reference
7. Python SDK Reference
8. Supported Dataset Formats
9. Breach Notifications
10. Running the Registry and Dashboard
11. Development Setup

---

1. What DCF Does

DCF solves a specific problem: when a producer pipeline changes a dataset (renames a column, changes a type, drops rows) the consumers of that dataset break silently. DCF gives you:

• A contract file (YAML) that declares what the dataset must look like
• An enforcement engine that validates the dataset against the contract at pipeline runtime
• A breach notifier that fires webhooks and emails when a contract is violated
• A registry that stores contract versions and breach history
• An observability dashboard to monitor all contracts across your data platform

---

2. Installation

Airflow workers and pipelines (core only)

pip install dcf

Installs: pydantic, pyyaml, pandas, pyarrow, httpx, sqlalchemy, typer. Does not install Streamlit or FastAPI — keeps your worker environment lean.

Registry server

pip install "dcf[registry]"

Streamlit dashboard

pip install "dcf[dashboard]"

Everything (development machine)

pip install "dcf[all]"

Install from local source

git clone <repo-url>
cd dcf
pip install -e .                    # core
pip install -e ".[all]"             # everything

---

3. Quick Start (5 minutes)

Prerequisites: Python 3.12+, Docker (for registry/dashboard)

1. Create a test dataset

# create_test_data.py
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

df = pd.DataFrame({
    "sale_id":  ["S001", "S002", "S003"],
    "amount":   [100.0, 250.0, 75.0],
    "currency": ["MYR", "MYR", "USD"],
})
pq.write_table(pa.Table.from_pandas(df), "/tmp/sales.parquet")
print("Dataset written.")

python create_test_data.py

2. Write a contract

# contracts/sales.yaml
apiVersion: datacontract/v1
kind: DataContract
metadata:
  name: daily_sales
  version: "1.0.0"
  owner:
    team: Sales Data Team
    email: [e-mail]
dataset:
  format: parquet
  location: /tmp/sales.parquet
on_breach: warn
schema:
  columns:
    - name: sale_id
      type: string
      nullable: false
    - name: amount
      type: float
      nullable: false
    - name: currency
      type: string
      allowed_values: [MYR, USD, SGD]
volume:
  min_rows: 1
quality:
  - column: sale_id
    max_null_percentage: 0.0
    max_duplicate_percentage: 0.0

3. Validate

dcf validate --contract contracts/sales.yaml
# ✓ daily_sales v1.0.0: COMPLIANT

That's it. No registry needed for basic validation.

---

4. Full Workflow

Step 1 — Write a Contract

Create a YAML file in your project repository (e.g. contracts/daily_sales.yaml). See Contract YAML Reference for all fields.

Contracts live in version control alongside pipeline code. When the schema changes, bump the version and commit.

---

Step 2 — Check the Contract

Validate the YAML syntax and Pydantic model without touching any data:

dcf check --contract contracts/daily_sales.yaml
# OK  daily_sales v1.0.0 — contract is valid

# If invalid:
# FAIL  dataset.format: Input should be 'parquet' or 'sql'

Run this in CI whenever a contract file changes:

# .github/workflows/validate-contracts.yml
on:
  push:
    paths: ["contracts/**/*.yaml"]
jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install dcf
      - run: dcf check --contract contracts/daily_sales.yaml

---

Step 3 — Start the Registry and Dashboard

The registry stores contract versions and breach history. The dashboard visualises them.

docker compose up -d

This starts:

• http://localhost:8000 — Registry REST API (FastAPI + PostgreSQL)
• http://localhost:8501 — Observability dashboard (Streamlit)
• PostgreSQL on port 5432 (internal)

Check it is running:

curl http://localhost:8000/health/
# {"status": "ok", "db": "connected"}

---

Step 4 — Publish the Contract

Register your contract with the registry. Do this once per contract version, typically in CI after merging a contract change.

dcf publish \
  --contract contracts/daily_sales.yaml \
  --registry-url http://localhost:8000
# Published daily_sales v1.0.0

The registry stores the full contract content, timestamps it, and marks it as the current version. Publishing a new version automatically marks the previous one as not current while preserving the full history.

View all published contracts:

dcf list --registry-url http://localhost:8000
#   daily_sales                               v1.0.0
#   customer_profiles                         v2.1.0

---

Step 5 — Validate in Your Pipeline

There are two ways to load the contract. Use whichever fits your environment:

	contract_path=	contract_name=
Where	local YAML file	fetched from registry at runtime
Best for	dev machine, CI scripts	Airflow workers, any remote runner
Requires registry?	No	Yes

Option A — local file (dev / CI)

from dcf import DataContractValidator, DataContractBreachError

write_data_to_parquet("/data/sales/daily.parquet")

result = DataContractValidator(
    contract_path="contracts/daily_sales.yaml",
    registry_url="http://localhost:8000",   # optional — enables breach history
).validate()

print(result.overall_status)   # COMPLIANT or BREACH
print(result.row_count)        # 48203

for clause in result.failed_clauses:
    print(clause.clause_type, clause.message)

Option B — fetch from registry by name (Airflow / remote workers)

result = DataContractValidator(
    contract_name="daily_sales",            # contract was published in Step 4
    registry_url="http://localhost:8000",   # required when using contract_name
).validate()

on_breach: warn (default) — validation result is returned, pipeline continues.
on_breach: fail — DataContractBreachError is raised, pipeline halts.

try:
    DataContractValidator(
        contract_name="daily_sales",
        registry_url="http://localhost:8000",
    ).validate()
except DataContractBreachError as e:
    print(f"Pipeline halted: {e}")
    print(f"Failed clauses: {len(e.result.failed_clauses)}")
    raise

---

Step 6 — Use in Airflow

Install DCF on your Airflow workers:

pip install dcf

Because the contract was already published to the registry in Step 4, workers do not need a local copy of the YAML file. Use contract_name= to fetch it directly from the registry at runtime:

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from dcf import DataContractValidator, DataContractBreachError
import os

REGISTRY_URL = os.environ.get("DCF_REGISTRY_URL", "http://dcf-registry:8000")

@dag(schedule_interval="@daily", start_date=days_ago(1), catchup=False)
def sales_pipeline():

    @task
    def extract_and_write():
        df = extract_from_source()
        df.to_parquet("/data/sales/daily.parquet")

    @task
    def validate():
        # No contract file needed on the worker — fetched from registry by name
        result = DataContractValidator(
            contract_name="daily_sales",
            registry_url=REGISTRY_URL,
        ).validate()
        return result.overall_status.value   # XCom-able

    @task
    def validate_strict():
        try:
            DataContractValidator(
                contract_name="daily_sales",
                registry_url=REGISTRY_URL,
            ).validate()
        except DataContractBreachError as e:
            raise Exception(f"Data quality check failed: {e}") from e

    extract_and_write() >> validate()

dag = sales_pipeline()

How it works:

Airflow worker                         Registry (already running)
──────────────                         ──────────────────────────
DataContractValidator(
  contract_name="daily_sales",  ──GET /contracts/daily_sales──▶  returns contract JSON
  registry_url=...,             ◀─────────────────────────────   (published in Step 4)
)
  │
  ├── reads dataset from contract.dataset.location
  ├── runs all validators
  ├── if BREACH → fires notifications
  └── POST /validation-results  ──────────────────────────────▶  stores breach history

Two ways to load a contract — summary:

# Option A: from local file (dev machine, CI)
DataContractValidator(contract_path="contracts/daily_sales.yaml", ...)

# Option B: from registry by name (Airflow workers — no local file needed)
DataContractValidator(contract_name="daily_sales", registry_url=REGISTRY_URL, ...)

---

Step 7 — Observe Breaches

Dashboard — open http://localhost:8501 in a browser.

Page	What you see
Overview	Total contracts, compliant count, breach count, recent breach events
Contract Detail	Full contract metadata, validation history chart
Breach History	Filterable table of all validation results
Discovery	Searchable list of all registered contracts

CLI history

dcf history \
  --name daily_sales \
  --registry-url http://localhost:8000 \
  --limit 20
# ✓ 2026-06-08T02:00:00  COMPLIANT
# ✗ 2026-06-07T02:00:00  BREACH
# ✓ 2026-06-06T02:00:00  COMPLIANT

Registry API directly

# All contracts
curl http://localhost:8000/contracts/

# Specific contract
curl http://localhost:8000/contracts/daily_sales

# Recent validation results
curl "http://localhost:8000/validation-results/?contract_name=daily_sales&limit=10"

The registry exposes interactive API docs at http://localhost:8000/docs.

---

5. Contract YAML Reference

apiVersion: datacontract/v1          # required — must be exactly this value
kind: DataContract                   # required — must be exactly this value

metadata:
  name: daily_sales                  # required — unique identifier, lowercase_underscores
  version: "1.0.0"                   # required — semantic version
  description: >                     # optional
    Human-readable description of the dataset.
  owner:
    team: Sales Data Team            # required
    email: [e-mail]         # required — breach emails go here
  tags: [finance, daily]             # optional

dataset:
  format: parquet                    # required — parquet | sql
  location: /data/sales/daily/       # required for parquet

  # For SQL datasets:
  # format: sql
  # connection_string: postgresql://user:pass@host:5432/db
  # table_name: daily_sales
  # partition_column: sale_date      # optional — used for freshness check

on_breach: warn                      # required — warn | fail

consumers:                           # optional — teams that depend on this dataset
  - team: Finance Reporting Team
    email: [e-mail]
  - team: ML Platform Team
    email: [e-mail]

notifications:                       # optional
  webhook:
    url: https://hooks.slack.com/services/YOUR/WEBHOOK/URL
    headers:
      Authorization: Bearer mytoken  # optional
  email:
    smtp_host: smtp.example.com
    smtp_port: 587
    smtp_user: [e-mail]
    smtp_password_env: SMTP_PASSWORD # name of env var — never put passwords in YAML
    recipients:
      - [e-mail]

schema:                              # optional
  enforce_no_extra_columns: false    # if true, extra columns trigger a breach
  columns:
    - name: sale_id
      type: string                   # string | integer | float | boolean | date | timestamp | decimal
      nullable: false                # default: true
      description: Unique sale ID    # optional
      allowed_values:                # optional — breach if any value is outside this list
        - SALE
        - REFUND

freshness:                           # optional
  max_age_hours: 25                  # breach if dataset is older than this
  check_column: sale_date            # optional — use max(sale_date) instead of file mtime

volume:                              # optional
  min_rows: 1000
  max_rows: 10000000

quality:                             # optional — column-level rules
  - column: sale_id
    max_null_percentage: 0.0         # 0.0 = no nulls allowed
    max_duplicate_percentage: 0.0    # 0.0 = must be unique
  - column: amount
    min_value: 0.01
    max_value: 9999999.0
  - column: currency
    max_null_percentage: 1.0

Column types

Type	Matches
string	object, StringDtype
integer	int32, int64
float	float32, float64
boolean	bool
date	datetime64
timestamp	datetime64 with timezone
decimal	float (exact decimal)

---

6. CLI Reference

dcf check     --contract PATH
dcf publish   --contract PATH  --registry-url URL
dcf validate  --contract PATH  [--registry-url URL]  [--output text|json]
dcf list      --registry-url URL
dcf history   --name NAME  --registry-url URL  [--limit N]

Command	What it does
check	Parse and validate the contract YAML. No dataset access. Safe to run in CI.
publish	POST the contract to the registry. Run once per contract version.
validate	Read the dataset, run all checks, print result. Exits 1 on breach.
list	Print all current contracts registered.
history	Print recent validation results for a named contract.

JSON output (useful for CI)

dcf validate --contract contracts/sales.yaml --output json
# {
#   "status": "BREACH",
#   "row_count": 8420,
#   "failed_clauses": [
#     {"clause_type": "schema.column_exists", "clause_target": "sale_id", "message": "..."}
#   ]
# }
echo $?   # 1 on breach, 0 on compliant

---

7. Python SDK Reference

DataContractValidator

Exactly one of contract_path or contract_name must be provided.

from dcf import DataContractValidator, DataContractBreachError

# Option A — from a local file
validator = DataContractValidator(
    contract_path="contracts/sales.yaml",   # path to contract YAML file
    registry_url="http://localhost:8000",   # optional — enables breach history posting
    extra_validators=[MyCustomValidator()], # optional — plugin validators
    notifiers=[],                           # optional — [] disables all notifications
)

# Option B — fetch from registry by name (no local file needed)
validator = DataContractValidator(
    contract_name="daily_sales",            # name as published via dcf publish
    registry_url="http://localhost:8000",   # required when using contract_name
)

result = validator.validate()

ValidationResult

result.overall_status        # OverallStatus.COMPLIANT | BREACH | ERROR
result.is_breach             # bool
result.row_count             # int
result.failed_clauses        # List[ClauseResult]
result.clause_results        # List[ClauseResult] — all clauses including passing

for c in result.failed_clauses:
    c.clause_type     # e.g. "schema.column_exists"
    c.clause_target   # column name or None
    c.expected        # what the contract declared
    c.observed        # what the engine measured
    c.message         # human-readable explanation

validate_dataframe() — for testing

Skip the storage read step and pass a DataFrame directly:

from dcf.engine import validate_dataframe
from dcf.models.contract import DataContract
import pandas as pd

df = pd.DataFrame({"id": ["A", "B"], "amount": [10.0, 20.0]})
result = validate_dataframe(df, contract)

Custom validator (plugin)

from dcf.validators.base import Validator
from dcf.models.result import ClauseResult, ClauseStatus

class MyCustomValidator(Validator):
    def validate(self, df, contract, reader_last_modified):
        ok = df["amount"].sum() > 0
        return [ClauseResult(
            clause_type="custom.total_amount_positive",
            clause_target="amount",
            status=ClauseStatus.PASS if ok else ClauseStatus.FAIL,
            expected="> 0",
            observed=str(df["amount"].sum()),
            message="" if ok else "Total amount must be positive",
        )]

result = DataContractValidator(
    "contracts/sales.yaml",
    extra_validators=[MyCustomValidator()],
).validate()

---

8. Supported Dataset Formats

Parquet (local or S3)

dataset:
  format: parquet
  location: /data/sales/daily.parquet      # local path
  # location: s3://my-bucket/sales/daily/  # S3 (requires s3fs)

SQL (via SQLAlchemy)

dataset:
  format: sql
  connection_string: postgresql://user:pass@localhost:5432/mydb
  table_name: daily_sales
  partition_column: sale_date              # optional — for freshness checks

Supported SQL databases: PostgreSQL, MySQL, SQLite, and any database with a SQLAlchemy driver.

---

9. Breach Notifications

Add a notifications block to your contract YAML. Notifications fire only on BREACH — not on COMPLIANT runs.

Slack webhook

notifications:
  webhook:
    url: https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK

Payload sent:

{
  "event": "DATA_CONTRACT_BREACH",
  "contract_name": "daily_sales",
  "contract_version": "1.0.0",
  "validated_at": "2026-06-08T02:15:00Z",
  "row_count": 8420,
  "failed_clauses": [...],
  "on_breach": "warn"
}

Email (SMTP)

notifications:
  email:
    smtp_host: smtp.gmail.com
    smtp_port: 587
    smtp_user: [e-mail]
    smtp_password_env: SMTP_PASSWORD    # set this env var on the worker
    recipients:
      - [e-mail]
      - [e-mail]

Set the password as an environment variable — never put it in the YAML file.

Disable notifications (useful in tests or local dev)

DataContractValidator(contract_path="contracts/sales.yaml", notifiers=[]).validate()
# or
DataContractValidator(contract_name="daily_sales", registry_url=REGISTRY_URL, notifiers=[]).validate()

---

10. Running the Registry and Dashboard

Start everything

docker compose up -d

Services

Service	URL	Purpose
Registry API	http://localhost:8000	REST API for contracts and breach history
API Docs	http://localhost:8000/docs	Interactive Swagger UI
Dashboard	http://localhost:8501	Observability UI
PostgreSQL	localhost:5432	Persistent storage

Stop and remove

docker compose down           # stop, keep data
docker compose down -v        # stop, delete all data

Environment variables

Variable	Default	Purpose
DATABASE_URL	sqlite:///./dcf_registry.db	Registry database connection
REGISTRY_URL	http://localhost:8000	Dashboard → registry URL
SMTP_PASSWORD	—	Email notifier password

---

11. Development Setup

git clone <repo-url>
cd dcf

# Install with all extras + dev tools
pip install "uv"
uv sync

# Run all tests
uv run pytest

# Run unit tests only (no Docker needed)
uv run pytest tests/unit/ -v

# Run integration tests (no Docker needed — uses SQLite)
uv run pytest tests/integration/ -v

# Run with coverage report
uv run pytest --cov=dcf --cov-report=html:htmlcov

# Start the registry locally (SQLite, no Docker)
uv run uvicorn registry.main:app --reload --port 8000

# Start the dashboard
uv run streamlit run dashboard/app.py --server.port 8501

# Check a contract
uv run dcf check --contract contracts/example_transactions.yaml

Project structure

dcf/
├── dcf/                    # Core Python package — install this on Airflow workers
│   ├── models/             # Pydantic contract models + validation result dataclasses
│   ├── readers/            # ParquetReader, SQLReader
│   ├── validators/         # SchemaValidator, FreshnessValidator, VolumeValidator, QualityValidator
│   ├── notifiers/          # WebhookNotifier, EmailNotifier
│   ├── engine.py           # Orchestrates readers + validators → ValidationResult
│   ├── sdk.py              # DataContractValidator — main public API
│   ├── cli.py              # dcf CLI commands
│   └── registry_client.py # HTTP client for posting to registry
├── registry/               # FastAPI registry service (run via Docker or uvicorn)
├── dashboard/              # Streamlit dashboard
├── tests/
│   ├── unit/               # Validator tests — no external dependencies
│   ├── integration/        # Engine + registry API tests — SQLite, no Docker
│   ├── fixtures/contracts/ # Sample contract YAML files
│   └── conftest.py         # Shared fixtures: make_contract(), make_transactions_df(), etc.
├── contracts/              # Example contract files
├── docker-compose.yml
└── pyproject.toml

---

DCF v1.0.0 — Faizal — MMU MCS Software Engineering — 2026