datacontract-framework 1.1.0

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

datacontract-framework

A lightweight Python library for defining, enforcing, and monitoring data quality contracts on batch datasets. Built for data engineering pipelines — works standalone, in Airflow, or any Python environment.

Published on PyPI: pip install datacontract-framework

---

What it does

When a producer pipeline changes a dataset (renames a column, drops rows, adds bad values), downstream consumers break silently. datacontract-framework 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 registry that stores contract versions and validation history
• A CLI for manual validation and contract management
• A dashboard to monitor all contracts across your data platform

---

Features

Validation Rules

Feature	What it checks
Schema — column existence	Every declared column is present in the dataset
Schema — column types	Column dtype matches declared type (string, integer, float, boolean, date, timestamp)
Schema — nullable	Non-nullable columns have zero null values
Schema — allowed values	Column contains only the declared set of allowed values
Schema — no extra columns	Dataset has no undeclared columns (optional, off by default)
Freshness	Dataset was updated within max_age_hours; uses file mtime or max(check_column)
Volume	Row count is within min_rows / max_rows bounds
Quality — null rate	Column null percentage does not exceed max_null_percentage
Quality — duplicate rate	Column duplicate percentage does not exceed max_duplicate_percentage
Quality — value range	Column values are within min_value / max_value bounds

Dataset Formats

Format	How
Parquet	Local path or S3 via pyarrow
SQL	Any SQLAlchemy-supported database (PostgreSQL, MySQL, SQLite) via table_name + connection_string

Breach Modes

Mode	Behaviour
on_breach: warn	Returns result with is_breach=True, pipeline continues
on_breach: fail	Raises DataContractBreachError, pipeline halts

Contract Loading

Method	When to use
contract_path="contracts/sales.yaml"	Dev machine, CI — file is local
contract_name="daily_sales" + registry_url=...	Airflow workers, remote runners — no local file needed

Notifications

• Webhook — POST JSON breach payload to any URL (Slack, Teams, PagerDuty)
• Email — SMTP with configurable recipients; password stored in env var, never in YAML

Registry

• REST API (FastAPI) — publish contracts, fetch by name, list versions, store validation results
• PostgreSQL backend for production; SQLite for local dev
• Interactive API docs at /docs

Observability Dashboard

• FastAPI + Jinja2 + Tailwind (CDN, no build step) — overview of all contracts, compliant vs breach counts, per-contract validation history, breach history with status filters, contract discovery/search

CLI

• dcf check — parse and validate YAML syntax without touching data (CI-safe)
• dcf publish — register a contract version
• dcf validate — run full validation, exit 1 on breach (CI-friendly)
• dcf list — list all current contracts in registry
• dcf history — show recent validation runs for a contract

Developer Experience

• validate_dataframe(df, contract) — skip storage reads in unit tests, pass a DataFrame directly
• Injectable _http_client and _registry_client — test the full SDK without a real server
• Custom validator plugin API — extend with your own business rules
• Split dependencies — pip install datacontract-framework (core only) keeps Airflow worker environments lean

---

Installation

pip install datacontract-framework                   # core — Airflow workers, pipelines
pip install "datacontract-framework[registry]"       # + registry server
pip install "datacontract-framework[all]"            # everything

---

Quick Start

1. Write a contract

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

2. Validate

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

# On breach:
# ✗ daily_sales v1.0.0: BREACH
# Failed clauses:
#   - [schema.allowed_values] [currency_code] Contains values not in allowed list: ['JPY']

---

Workflow

Step 1 — Check contract syntax

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

Step 2 — Start the registry

docker compose up -d

• Registry API: http://localhost:8000
• Dashboard: http://localhost:8501

Step 3 — Publish the contract

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

Step 4 — Validate in your pipeline

From a local file (dev / CI):

from dcf import DataContractValidator, DataContractBreachError

result = DataContractValidator(
    contract_path="contracts/sales.yaml",
    registry_url="http://localhost:8000",
).validate()

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

From the registry by name (Airflow workers — no local file needed):

result = DataContractValidator(
    contract_name="daily_sales",
    registry_url="http://dcf-registry:8000",
).validate()

Step 5 — Use in Airflow

from airflow.sdk import dag, task
from dcf import DataContractValidator
import os

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

@dag(schedule="@daily", ...)
def sales_pipeline():

    @task
    def extract_and_load() -> int:
        # write dataset to /data/sales/daily.parquet
        ...

    @task
    def validate(row_count: int) -> str:
        result = DataContractValidator(
            contract_name="daily_sales",   # fetched from registry — no local file
            registry_url=REGISTRY_URL,
            notifiers=[],
        ).validate()

        if result.is_breach:
            raise ValueError(f"Contract breach — pipeline halted")

        return result.overall_status.value

    @task
    def transform(status: str) -> None:
        ...  # only runs when validation passes

    rows = extract_and_load()
    status = validate(rows)
    transform(status)

On breach: validate raises → Airflow marks it FAILED → transform is skipped — bad data never reaches downstream consumers.

---

Contract YAML Reference

apiVersion: datacontract/v1
kind: DataContract

metadata:
  name: daily_sales          # unique identifier
  version: "1.0.0"           # semantic version
  owner:
    team: Data Engineering
    email: [e-mail]
  tags: [finance, daily]

dataset:
  format: parquet             # parquet | sql
  location: /data/sales/daily.parquet

  # SQL datasets:
  # format: sql
  # connection_string: postgresql://user:pass@host:5432/db
  # table_name: daily_sales

on_breach: warn               # warn | fail

schema:
  enforce_no_extra_columns: false
  columns:
    - name: sale_id
      type: string            # string | integer | float | boolean | date | timestamp
      nullable: false
      allowed_values: [SALE, REFUND]

freshness:
  max_age_hours: 25
  check_column: sale_date     # optional — uses max(column) instead of file mtime

volume:
  min_rows: 1000
  max_rows: 10000000

quality:
  - column: sale_id
    max_null_percentage: 0.0
    max_duplicate_percentage: 0.0
  - column: amount
    min_value: 0.01
    max_value: 9999999.0

notifications:
  webhook:
    url: https://hooks.slack.com/services/YOUR/WEBHOOK/URL
  email:
    smtp_host: smtp.example.com
    smtp_port: 587
    smtp_user: [e-mail]
    smtp_password_env: SMTP_PASSWORD
    recipients:
      - [e-mail]

---

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]

---

Python SDK Reference

DataContractValidator

from dcf import DataContractValidator, DataContractBreachError

# Option A — from local file
validator = DataContractValidator(
    contract_path="contracts/sales.yaml",
    registry_url="http://localhost:8000",   # optional — enables breach history
    extra_validators=[MyValidator()],       # optional plugins
    notifiers=[],                           # [] disables notifications
)

# Option B — from registry by name (Airflow / remote workers)
validator = DataContractValidator(
    contract_name="daily_sales",
    registry_url="http://localhost:8000",
)

result = validator.validate()

ValidationResult

result.overall_status      # OverallStatus.COMPLIANT | BREACH | ERROR
result.is_breach           # bool
result.row_count           # int
result.failed_clauses      # List[ClauseResult]

for c in result.failed_clauses:
    print(c.clause_type)   # e.g. "schema.allowed_values"
    print(c.clause_target) # column name
    print(c.message)       # human-readable explanation

validate_dataframe() — for unit testing

from dcf.engine import validate_dataframe
import pandas as pd

df = pd.DataFrame({"sale_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 MyValidator(Validator):
    def validate(self, df, contract, reader_last_modified):
        ok = df["amount"].sum() > 0
        return [ClauseResult(
            clause_type="custom.positive_total",
            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",
        )]

DataContractValidator(
    contract_path="contracts/sales.yaml",
    extra_validators=[MyValidator()],
).validate()

---

Development Setup

git clone <repo-url>
cd dcf

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

# Run tests
uv run pytest

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

# Integration tests (SQLite, no Docker)
uv run pytest tests/integration/ -v

# Start registry locally
uv run uvicorn registry.main:app --reload --port 8000

# Start dashboard
uv run uvicorn dashboard.main:app --reload --port 8501

# Start everything (registry + dashboard + postgres)
docker compose up -d

Project structure

dcf/
├── dcf/                    # Core package — install this on Airflow workers
│   ├── models/             # Contract and result Pydantic models
│   ├── readers/            # ParquetReader, SQLReader
│   ├── validators/         # Schema, Freshness, Volume, Quality validators
│   ├── notifiers/          # Webhook, Email notifiers
│   ├── engine.py           # Orchestrates readers + validators
│   ├── sdk.py              # DataContractValidator — main public API
│   ├── cli.py              # dcf CLI
│   └── registry_client.py  # HTTP client for the registry
├── registry/               # FastAPI registry service
├── dashboard/              # FastAPI + Jinja2 + Tailwind observability dashboard
├── lab/                    # End-to-end Docker test lab
├── tests/
│   ├── unit/               # Validator unit tests
│   ├── integration/        # Engine + registry API tests
│   └── fixtures/           # Sample contract YAML files
├── contracts/              # Example contracts
├── docker-compose.yml
└── pyproject.toml

---

datacontract-framework v1.1.0 — Faizal Azman — MMU MCS Software Engineering — 2026