Metadata-Version: 2.4
Name: datacontract-framework
Version: 1.0.0
Summary: datacontract-framework — Lightweight Data Contract Framework for automated data quality enforcement in lakehouse pipelines
Author-email: Faizal Azman <faizalazman88@gmail.com>
License-Expression: MIT
Project-URL: Source Code, https://github.com/faizalazman/datacontract-framework
Project-URL: Bug Tracker, https://github.com/faizalazman/datacontract-framework/issues
Keywords: data-contracts,data-quality,data-engineering,airflow,lakehouse
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Database
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.28
Requires-Dist: pandas>=2.0
Requires-Dist: pyarrow>=14.0
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: typer>=0.9
Provides-Extra: registry
Requires-Dist: alembic>=1.18; extra == "registry"
Requires-Dist: fastapi>=0.136; extra == "registry"
Requires-Dist: psycopg2-binary>=2.9; extra == "registry"
Requires-Dist: uvicorn[standard]>=0.29; extra == "registry"
Provides-Extra: dashboard
Requires-Dist: streamlit>=1.33; extra == "dashboard"
Requires-Dist: httpx>=0.28; extra == "dashboard"
Provides-Extra: all
Requires-Dist: datacontract-framework[registry]; extra == "all"
Requires-Dist: datacontract-framework[dashboard]; extra == "all"
Dynamic: license-file

# 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](#1-what-dcf-does)
2. [Installation](#2-installation)
3. [Quick Start (5 minutes)](#3-quick-start-5-minutes)
4. [Full Workflow](#4-full-workflow)
   - [Step 1 — Write a Contract](#step-1--write-a-contract)
   - [Step 2 — Check the Contract](#step-2--check-the-contract)
   - [Step 3 — Start the Registry and Dashboard](#step-3--start-the-registry-and-dashboard)
   - [Step 4 — Publish the Contract](#step-4--publish-the-contract)
   - [Step 5 — Validate in Your Pipeline](#step-5--validate-in-your-pipeline)
   - [Step 6 — Use in Airflow](#step-6--use-in-airflow)
   - [Step 7 — Observe Breaches](#step-7--observe-breaches)
5. [Contract YAML Reference](#5-contract-yaml-reference)
6. [CLI Reference](#6-cli-reference)
7. [Python SDK Reference](#7-python-sdk-reference)
8. [Supported Dataset Formats](#8-supported-dataset-formats)
9. [Breach Notifications](#9-breach-notifications)
10. [Running the Registry and Dashboard](#10-running-the-registry-and-dashboard)
11. [Development Setup](#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)

```bash
pip install dcf
```

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

### Registry server

```bash
pip install "dcf[registry]"
```

### Streamlit dashboard

```bash
pip install "dcf[dashboard]"
```

### Everything (development machine)

```bash
pip install "dcf[all]"
```

### Install from local source

```bash
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**

```python
# 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.")
```

```bash
python create_test_data.py
```

**2. Write a contract**

```yaml
# contracts/sales.yaml
apiVersion: datacontract/v1
kind: DataContract
metadata:
  name: daily_sales
  version: "1.0.0"
  owner:
    team: Sales Data Team
    email: you@example.com
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**

```bash
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](#5-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:

```bash
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:

```yaml
# .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.

```bash
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:

```bash
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.

```bash
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:

```bash
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)**

```python
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)**

```python
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.

```python
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:

```bash
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:

```python
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:**

```python
# 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**

```bash
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**

```bash
# 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

```yaml
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: sales@example.com         # 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: finance@example.com
  - team: ML Platform Team
    email: mlplatform@example.com

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: alerts@example.com
    smtp_password_env: SMTP_PASSWORD # name of env var — never put passwords in YAML
    recipients:
      - data-team@example.com

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)**

```bash
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.

```python
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`

```python
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:

```python
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)

```python
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)

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

### SQL (via SQLAlchemy)

```yaml
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

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

Payload sent:
```json
{
  "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)

```yaml
notifications:
  email:
    smtp_host: smtp.gmail.com
    smtp_port: 587
    smtp_user: alerts@example.com
    smtp_password_env: SMTP_PASSWORD    # set this env var on the worker
    recipients:
      - data-team@example.com
      - oncall@example.com
```

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

### Disable notifications (useful in tests or local dev)

```python
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

```bash
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

```bash
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

```bash
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*
