Data Quality Monitoring

Learn to monitor contract (row-based) data quality, track violations, and set up alerts. For pipeline (column/dataset-based) quality checks on ETL runs (e.g. row count, null rate), see Pipeline quality and Data quality: contract vs pipeline.

What You'll Learn

• Run quality checks on your data
• Calculate quality metrics
• Set up threshold alerts
• Track and query violations
• Generate quality reports
• Integrate with pipelines

Prerequisites

pip install pycharter

Part 1: Quick Start with Convenience Functions

The fastest way to run a quality check is the check_quality() function — pass a contract and data, get a report:

from pycharter import check_quality

contract = {
    "json_schema": {
        "version": "1.0.0",
        "properties": {
            "name": {"type": "string", "minLength": 1},
            "email": {"type": "string", "format": "email"},
            "age": {"type": "integer", "minimum": 0, "maximum": 150},
        },
        "required": ["name", "email"],
    }
}

data = [
    {"name": "Alice", "email": "alice@example.com", "age": 30},
    {"name": "Bob", "email": "invalid-email", "age": 25},
    {"name": "", "email": "charlie@example.com", "age": -5},
]

report = check_quality(contract=contract, data=data)

print(f"Quality Score: {report.quality_score.overall_score:.1f}/100")
print(f"Records: {report.record_count}")
print(f"Valid: {report.valid_count}")
print(f"Invalid: {report.invalid_count}")

Presets

QualityCheckOptions offers three presets for common scenarios:

from pycharter import check_quality, QualityCheckOptions

# basic() — metrics + violations, no profiling or thresholds (default)
report = check_quality(contract, data)

# strict() — everything on, including thresholds with defaults
report = check_quality(contract, data, options=QualityCheckOptions.strict())
if not report.passed:
    print(f"Threshold breaches: {report.threshold_breaches}")

# monitoring() — strict + skip-if-unchanged + dedup violations
report = check_quality(contract, data, options=QualityCheckOptions.monitoring())

Preset	Metrics	Violations	Profiling	Thresholds	Skip unchanged	Dedup
basic()	Yes	Yes	No	No	No	Yes
strict()	Yes	Yes	Yes	Yes (defaults)	No	Yes
monitoring()	Yes	Yes	Yes	Yes (defaults)	Yes	Yes

Contract target and SLA reports

pycharter.quality.contract_reporting.compute_contract_quality_report() produces a separate ContractQualityReport for a contract's declared targets and SLAs. It carries contract_namespace from contract metadata. Direct construction accepts contract_namespace as an optional keyword, defaulting to None for existing global reports; the original positional arguments retain their order. to_persistable() includes the namespace, but writing and querying reports remains the caller's responsibility.

Part 1.5: Running checks on a schedule

PyCharter does not run cron jobs for you. For hourly/daily checks, use cron, Airflow, or the validation worker (POST /api/v1/quality/jobs).

from pycharter import check_quality, QualityCheckOptions

report = check_quality(
    contract="contracts/user.yaml",
    data=fetch_production_data(),
    options=QualityCheckOptions.monitoring(),
)
if not report.passed:
    notify_team(report.threshold_breaches)

Worker (async):

from pycharter.worker import submit_job_sync

submit_job_sync(
    contract_name="user",
    contract_version="1.0.0",
    data_source="/data/users.jsonl",
    options={"check_thresholds": True, "skip_if_unchanged": True},
    idempotency_key="user-hourly-2026-06-04T12",
)

Full patterns (pipeline quality: true, contract schedule metadata, alerts): Periodic quality monitoring.

Standalone Data Profiling

Profile a dataset without a contract using profile_data():

from pycharter import profile_data

profile = profile_data(data)
print(f"Records: {profile['record_count']}")
print(f"Completeness: {profile['overall_stats']['completeness']:.1f}%")

for field, stats in profile["field_profiles"].items():
    print(f"  {field}: {stats['null_count']} nulls, {stats['distinct_count']} distinct")

Using the QualityCheck Class

For store-backed schemas, database persistence, or advanced options, use the QualityCheck class directly:

from pycharter import QualityCheck, SQLiteContractStore

store = SQLiteContractStore("quality.db")
store.connect()

schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 0, "maximum": 150},
    },
    "required": ["name", "email"],
}
store.store_json_schema("user", "1.0.0", schema)

check = QualityCheck(store=store)
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=[
        {"name": "Alice", "email": "alice@example.com", "age": 30},
        {"name": "Bob", "email": "invalid-email", "age": 25},
        {"name": "", "email": "charlie@example.com", "age": -5},
    ],
)

print(f"Quality Score: {report.quality_score.overall_score:.1f}/100")

There is also a convenience function for store-based checks:

from pycharter.quality import check_quality

report = check_quality("user@1.0.0", data, store=store)

Part 2: Quality Metrics

Understanding Metrics

Metric	Description	Formula
overall_score	Overall quality (0-100)	Weighted combination
violation_rate	% of records with errors	invalid / total
completeness	% of non-null required fields	filled / required
accuracy	% of valid values	valid_values / total_values
field_scores	Per-field quality scores	Varies by field

Accessing Metrics

report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=data,
    options=QualityCheckOptions(),
)

# Overall metrics
score = report.quality_score
print(f"Overall Score: {score.overall_score:.1f}")
print(f"Violation Rate: {score.violation_rate:.2%}")
print(f"Completeness: {score.completeness:.2%}")
print(f"Accuracy: {score.accuracy:.2%}")

# Record counts
print(f"Total Records: {report.record_count}")
print(f"Valid Records: {report.valid_count}")
print(f"Invalid Records: {report.invalid_count}")
print(f"Violations: {report.violation_count}")

# Field-level scores
for field, field_score in score.field_scores.items():
    print(f"  {field}: {field_score:.1f}")

Field Metrics

# Get detailed field metrics
for field_name, metrics in report.field_metrics.items():
    print(f"\n{field_name}:")
    print(f"  Null Count: {metrics.null_count}")
    print(f"  Violation Count: {metrics.violation_count}")
    print(f"  Total Count: {metrics.total_count}")
    print(f"  Violation Rate: {metrics.violation_rate:.2%}")
    print(f"  Completeness: {metrics.completeness:.2%}")

    if metrics.error_types:
        print(f"  Error Types: {metrics.error_types}")

Part 3: Quality Thresholds

Set alerts when quality drops below acceptable levels:

Defining Thresholds

from pycharter import QualityCheck, QualityCheckOptions, QualityThresholds

# Define thresholds
thresholds = QualityThresholds(
    min_overall_score=95.0,  # Alert if score < 95
    max_violation_rate=0.05,  # Alert if violations > 5%
    min_completeness=0.99,  # Alert if completeness < 99%
    min_accuracy=0.98,  # Alert if accuracy < 98%
)

# Run check with thresholds
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=data,
    options=QualityCheckOptions(check_thresholds=True, thresholds=thresholds),
)

# Check if thresholds passed
print(f"Passed: {report.passed}")
print(f"Threshold Breaches: {report.threshold_breaches}")

Handling Threshold Breaches

report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=data,
    options=QualityCheckOptions(check_thresholds=True, thresholds=thresholds),
)

if not report.passed:
    print("⚠️ Quality thresholds breached!")

    for breach in report.threshold_breaches:
        print(f"  - {breach}")

    # Take action
    send_alert(
        message=f"Data quality dropped to {report.quality_score.overall_score:.1f}",
        breaches=report.threshold_breaches,
    )

Per-Field Thresholds

from pycharter import QualityCheckOptions

options = QualityCheckOptions(
    calculate_metrics=True,
    record_violations=True,
    check_thresholds=True,
    thresholds=thresholds,
    field_thresholds={
        "email": {"max_violation_rate": 0.01},  # Strict for email
        "age": {"max_violation_rate": 0.10},  # Lenient for age
    },
)

report = check.run(
    contract_name="user", contract_version="1.0.0", data=data, options=options
)

Part 4: Violation Tracking

Track individual validation errors for analysis and remediation:

Recording Violations

from pycharter import QualityCheckOptions

options = QualityCheckOptions(
    record_violations=True,  # Enable violation tracking
    calculate_metrics=True,
)

report = check.run(
    contract_name="user", contract_version="1.0.0", data=data, options=options
)

print(f"Violations recorded: {report.violation_count}")

Querying Violations

Use ViolationTracker with a database session. The contract is identified by schema_id in the form "name:version" (e.g. "user:1.0.0"):

from pycharter.db.session import get_session
from pycharter.quality import ViolationTracker

session = get_session(database_url)
tracker = ViolationTracker(db_session=session)
violations = tracker.get_violations(
    schema_id="user:1.0.0",  # contract name : version
    status="open",  # open, resolved, ignored
    severity="critical",  # critical, warning, info
    limit=100,
)

for violation in violations:
    print(f"Record ID: {violation.record_id}")
    print(f"Field: {violation.field_violations}")
    print(f"Severity: {violation.severity}")
    print(f"Timestamp: {violation.timestamp}")
    print("---")

Violation Analysis

# Aggregate violations by field
from collections import Counter

violations = tracker.get_violations(schema_id="user:1.0.0")

field_counts = Counter()
error_types = Counter()

for v in violations:
    for field_error in v.field_violations:
        field_counts[field_error["field"]] += 1
        error_types[field_error["type"]] += 1

print("Violations by field:")
for field, count in field_counts.most_common():
    print(f"  {field}: {count}")

print("\nViolations by error type:")
for error_type, count in error_types.most_common():
    print(f"  {error_type}: {count}")

Resolving Violations

# Mark violation as resolved
tracker.resolve_violation(violation.id, resolved_by="alice@example.com")

# Resolve or ignore (implementation may offer status updates)
tracker.resolve_violation(violation.id, resolved_by="alice@example.com")

Part 5: Quality Reports

Basic Report

report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=data,
    options=QualityCheckOptions(),
)

# Access report data (report.schema_id is "name:version")
print(f"Contract: {report.schema_id}")
print(f"Timestamp: {report.check_timestamp}")
print(f"Duration: {report.metadata.get('duration_seconds', 'N/A')}s")

Exporting Reports

import json

# To dictionary
report_dict = {
    "schema_id": report.schema_id,
    "timestamp": report.check_timestamp,
    "quality_score": {
        "overall": report.quality_score.overall_score,
        "violation_rate": report.quality_score.violation_rate,
        "completeness": report.quality_score.completeness,
        "accuracy": report.quality_score.accuracy,
    },
    "record_counts": {
        "total": report.record_count,
        "valid": report.valid_count,
        "invalid": report.invalid_count,
    },
    "passed": report.passed,
    "threshold_breaches": report.threshold_breaches,
}

# To JSON
with open("quality_report.json", "w") as f:
    json.dump(report_dict, f, indent=2, default=str)

# To CSV (metrics history)
import csv

with open("quality_history.csv", "a", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(
        [
            report.check_timestamp,
            report.quality_score.overall_score,
            report.quality_score.violation_rate,
            report.record_count,
            report.valid_count,
        ]
    )

Historical Tracking

# Store quality metrics
metrics_id = store.store_quality_metrics(
    contract_name="user",
    contract_version="1.0.0",
    quality_score=report.quality_score,
    record_count=report.record_count,
    valid_count=report.valid_count,
    invalid_count=report.invalid_count,
    violation_count=report.violation_count,
)

# Query historical metrics
history = store.get_quality_metrics_history(
    contract_name="user",
    contract_version="1.0.0",
    limit=30,  # Last 30 checks
)

# Analyze trends
import statistics

scores = [m.overall_score for m in history]
print(f"Average Score: {statistics.mean(scores):.1f}")
print(f"Min Score: {min(scores):.1f}")
print(f"Max Score: {max(scores):.1f}")
print(f"Std Dev: {statistics.stdev(scores):.2f}")

Part 6: Pipeline Integration

Quality Gate Pattern

import asyncio
from pycharter import Pipeline, QualityCheck, QualityThresholds


async def run_pipeline_with_quality_gate():
    # Run ETL pipeline
    pipeline = Pipeline.from_dir("pipelines/users/")
    result = await pipeline.arun()

    if not result.success:
        raise Exception(f"Pipeline failed: {result.errors}")

    # Run quality check on output
    check = QualityCheck(store=store)
    report = check.run(
        contract_name="user",
        contract_version="1.0.0",
        data=read_output("output/users.json"),
        options=QualityCheckOptions(
            check_thresholds=True,
            thresholds=QualityThresholds(min_overall_score=95.0),
        ),
    )

    if not report.passed:
        # Quality gate failed - don't proceed
        raise Exception(f"Quality gate failed: {report.threshold_breaches}")

    # Quality passed - continue to next stage
    print(f"Quality gate passed with score {report.quality_score.overall_score:.1f}")


asyncio.run(run_pipeline_with_quality_gate())

Inline Validation with Metrics

from pycharter import Pipeline, CustomFunction, Validator

validator = Validator.from_file("contracts/user.yaml")
metrics = {"valid": 0, "invalid": 0, "errors": []}


def validate_with_metrics(records):
    valid_records = []
    for record in records:
        result = validator.validate(record)
        if result.is_valid:
            metrics["valid"] += 1
            valid_records.append(result.data.model_dump())
        else:
            metrics["invalid"] += 1
            metrics["errors"].extend(result.errors)
    return valid_records


pipeline = Pipeline(extractor) | CustomFunction(validate_with_metrics) | loader

result = asyncio.run(pipeline.run())

# Access metrics after run
print(f"Valid: {metrics['valid']}")
print(f"Invalid: {metrics['invalid']}")
print(f"Quality: {metrics['valid'] / (metrics['valid'] + metrics['invalid']):.2%}")

Scheduled Quality Checks

See Part 1.5 and the dedicated guide Periodic quality monitoring for cron, Airflow, worker jobs, and pipeline-integrated checks. You can still use the Python schedule library in a long-running process if you prefer that to system cron.

Part 7: Best Practices

1. Set Realistic Thresholds

# Start with lenient thresholds
initial_thresholds = QualityThresholds(min_overall_score=80.0, max_violation_rate=0.20)

# Tighten over time as data quality improves
production_thresholds = QualityThresholds(
    min_overall_score=95.0, max_violation_rate=0.05
)

2. Track Metrics Over Time

# Run check with record_violations and db_session to persist metrics
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=data,
    options=QualityCheckOptions(record_violations=True, calculate_metrics=True),
)
# Metrics and violations are stored when QualityCheck is created with db_session
print(f"Score: {report.quality_score.overall_score}")

3. Categorize Violations

# Use severity levels
def classify_violation(violation):
    critical_fields = ["email", "payment_id"]
    high_fields = ["name", "phone"]

    field = violation["field"]

    if field in critical_fields:
        return "critical"
    elif field in high_fields:
        return "high"
    else:
        return "medium"

4. Automate Remediation

def auto_remediate(violation):
    """Attempt automatic fixes for known issues."""

    if violation.error_type == "string_pattern_mismatch":
        if violation.field == "phone":
            # Auto-fix: strip non-digits
            return fix_phone_number(violation.value)

    return None  # Cannot auto-fix

Exercises

1. Basic Check: Run a quality check on a dataset and print all metrics.

2. Thresholds: Set up thresholds and implement an alert system for breaches.

3. Violation Analysis: Query violations and create a summary report by field and error type.

4. Pipeline Integration: Add a quality gate to an ETL pipeline that stops if quality drops below 90%.

5. Historical Tracking: Store quality metrics over time and plot a trend chart.

Next Steps

• Contract Store Tutorial - Store quality metrics
• REST API Tutorial - Access quality via API
• API Reference: QualityCheck - Complete documentation