Data Quality Checks in Python: Catch Bad Data Before Your Users Do

Every data engineer learns the same lesson: pipelines rarely fail loudly. They succeed while loading garbage — nulls where IDs should be, duplicates from a retried batch, yesterday’s file loaded twice. Quality checks turn silent corruption into loud, fixable failures.

This is the framework I drop into Python jobs before a warehouse load. It is not a philosophy deck. It is four checks, a fail-vs- quarantine rule, and an exit into dbt or Great Expectations when the YAML file gets longer than the pipeline.

flowchart LR
  extract[Extract] --> transform[Transform]
  transform --> dq[Quality checks]
  dq -->|pass| load[Load]
  dq -->|fail| alert[Alert + stop]
  dq -->|messy| quarantine[Quarantine table]

Where the check lives

Put checks after transform, before load. Data that fails never reaches the table the BI tool reads.

• In the extract you can assert “file arrived and is parseable.” That is a gate, not a grain test.
• In the warehouse (dbt tests) you catch what Python missed and what other jobs wrote. Both layers is correct. Only-after-load is how executives see the bad number first.

If the pipeline is already dbt-shaped, start with the dbt tutorial tests and use this Python module for the pandas / DuckDB / API path that never hits dbt.

The four checks that catch most problems

For any batch, in descending order of value:

1. Volume — did roughly the expected number of rows arrive?
2. Uniqueness — is the business key actually unique?
3. Completeness — are critical columns non-null?
4. Freshness — is the newest timestamp recent enough?

Range / enum checks are the fifth that pays for itself on amounts and statuses. I still lead with the four because they fail on almost every incident I have had to explain.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

import pandas as pd


class DataQualityError(Exception):
    pass


@dataclass(frozen=True)
class CheckConfig:
    table: str
    key: str
    required: tuple[str, ...]
    min_rows: int
    max_rows: int | None = None
    amount_col: str | None = "amount"
    ts_col: str | None = "updated_at"
    max_lag: timedelta = timedelta(hours=26)


def run_checks(df: pd.DataFrame, cfg: CheckConfig) -> None:
    failures: list[str] = []

    n = len(df)
    if n < cfg.min_rows:
        failures.append(f"only {n} rows (expected >= {cfg.min_rows})")
    if cfg.max_rows is not None and n > cfg.max_rows:
        failures.append(f"{n} rows (expected <= {cfg.max_rows})")

    if cfg.key not in df.columns:
        raise DataQualityError(f"{cfg.table}: missing key column {cfg.key}")

    dupes = int(df[cfg.key].duplicated().sum())
    if dupes:
        failures.append(f"{dupes} duplicate {cfg.key} values")

    for col in cfg.required:
        nulls = int(df[col].isna().sum())
        if nulls:
            failures.append(f"{nulls} nulls in {col}")

    if cfg.amount_col and cfg.amount_col in df.columns:
        bad = int((df[cfg.amount_col] < 0).sum())
        if bad:
            failures.append(f"{bad} negative {cfg.amount_col} values")

    if cfg.ts_col and cfg.ts_col in df.columns and n:
        newest = pd.to_datetime(df[cfg.ts_col], utc=True).max()
        now = datetime.now(timezone.utc)
        if newest < now - cfg.max_lag:
            failures.append(
                f"newest {cfg.ts_col} is {newest.isoformat()} "
                f"(lag > {cfg.max_lag})"
            )

    if failures:
        raise DataQualityError(f"{cfg.table}: " + "; ".join(failures))

cfg = CheckConfig(
    table="orders",
    key="order_id",
    required=("order_id", "customer_id", "amount"),
    min_rows=1_000,
    max_rows=5_000_000,
)
run_checks(orders_df, cfg)

Volume uses a band, not an exact count. Tuesday after a holiday is not an incident. Zero rows on a weekday is. Tune min_rows from a week of successful runs, then revisit when the business changes — not every time Slack is noisy.

Fail hard or quarantine?

Two valid strategies, chosen per dataset, not per mood:

• Fail the pipeline when downstream correctness matters more than freshness (finance, billing, anything that books revenue). A stale-but-correct dashboard beats a fresh wrong one.
• Quarantine messy rows when partial data is useful: write failures to a rejected_rows table with a reason, load the rest, and alert. Support tickets and clickstream often belong here. Invoice lines usually do not.

from datetime import date

today = date.today().isoformat()

key_dup = orders_df["order_id"].duplicated(keep=False)
neg = orders_df["amount"] < 0
bad = key_dup | neg

rejected = orders_df.loc[bad].copy()
rejected["reject_reason"] = None
rejected.loc[neg, "reject_reason"] = "negative_amount"
rejected.loc[key_dup & ~neg, "reject_reason"] = "duplicate_order_id"

rejected.to_parquet(f"quarantine/orders/dt={today}.parquet")
clean = orders_df.loc[~bad].copy()
run_checks(clean, cfg)  # remaining contract must still hold

Never quarantine volume or freshness failures by dropping rows. Those are batch-level. You cannot “reject” the fact that Monday’s file never arrived.

flowchart TD
  batch[Transformed frame] --> gate{run_checks}
  gate -->|pass| load[Load gold]
  gate -->|row mess and useful rest| q[Parquet quarantine + load clean]
  gate -->|volume / freshness / schema| page[Raise and page]

Checks sit between transform and load. Quarantine is a fork, not a way to hide a missing file.

# dq/orders.yaml — bands from a week of good runs, not from Slack
table: orders
key: order_id
required:
  - order_id
  - customer_id
  - amount
min_rows: 1000
max_rows: 5000000
max_lag_hours: 26
row_failures: quarantine
batch_failures: fail
quarantine_uri: s3://lake/quarantine/orders/dt={run_date}/

Load that file in the job. Changing min_rows in chat is how the check becomes decoration.

Schema is a check

Types and required columns are quality. A renamed API field that becomes all-null will pass a naive amount > 0 check on an empty frame if you are not careful.

REQUIRED_DTYPES = {
    "order_id": "string",
    "customer_id": "string",
    "amount": "float64",
}


def assert_schema(df: pd.DataFrame) -> None:
    missing = set(REQUIRED_DTYPES) - set(df.columns)
    if missing:
        raise DataQualityError(f"missing columns: {sorted(missing)}")
    for col, dtype in REQUIRED_DTYPES.items():
        if str(df[col].dtype) != dtype:
            # pandas dtypes are messy; normalize if you use Arrow
            raise DataQualityError(
                f"{col} has dtype {df[col].dtype}, expected {dtype}"
            )

Pandera does this with less boilerplate once the team is ready (see below). Hand-rolled is fine for one job.

Incremental loads: check the slice, then the table

A daily incremental job should check today’s slice (volume, key unique within the slice, nulls) and periodically check the target table (key unique globally, freshness).

A slice can be unique while the MERGE target is not — leftover duplicates from last month. Warehouse-side tests catch that. Python-only checks on the dataframe you are about to load will not.

-- warehouse companion (BigQuery / Snowflake / DuckDB)
select order_id, count(*) as n
from warehouse.orders
group by 1
having count(*) > 1
limit 20;

If that query returns rows, the Python job “passed” and you still have a bad grain. Pair the layers.

Wiring it so someone wakes up

A failing check that writes to a log nobody reads is theater.

import logging
import os

import requests

log = logging.getLogger("dq")


def notify(message: str) -> None:
    webhook = os.environ.get("DQ_SLACK_WEBHOOK")
    if webhook:
        requests.post(webhook, json={"text": message}, timeout=10)
    log.error(message)


def guarded_load(df: pd.DataFrame, cfg: CheckConfig) -> None:
    try:
        run_checks(df, cfg)
    except DataQualityError as exc:
        notify(f":rotating_light: DQ failed — {exc}")
        raise
    load_to_warehouse(df)

Page on contract breaks. Ticket or daily digest on quarantine volume crossing a threshold. Recurring failures are upstream bugs, not a reason to widen min_rows until the check is silent.

In Airflow, raise DataQualityError and let the task fail. Do not pass after a Slack message.

When the file outgrows a file

Hand-rolled checks are the right start. Graduate when:

• The same rules are copy-pasted across three repos.
• Analysts need to see pass/fail history without reading CI logs.
• You want data docs attached to the rules.

Options that have worked in real teams:

• dbt tests — if the data already lands in a warehouse. Natural home for grain and relationships. Start here if you have dbt.
• Great Expectations / Soda — richer expectation suites, profiling, and reports for pipelines that are not dbt.
• Pandera — DataFrame schemas in Python, good inside libraries and services.

I do not start with GX on a 200-line extractor. I start with run_checks. I also do not keep a 400-line checks.py that is secretly a framework. That is the signal to pick one of the above.

Pitfalls

• Exact row counts. They flap. Use bands and day-of-week awareness if you must, or compare to a moving median.
• Checking only not-null on the key and ignoring uniqueness. Retries produce duplicates more often than null keys.
• Quarantining without an owner. The parquet pile becomes a landfill. Review it or delete the feature.
• assert instead of a real exception. -O strips asserts. Use DataQualityError.
• Sampling 1% and declaring the batch clean. Samples miss the one null customer_id in the last partition.
• Validating after to_sql(if_exists="append"). The damage is already in the table.

When NOT to add (more) checks

• You cannot name who gets paged. Fix ownership first.
• The “bad” values are the product (cancelled orders, $0 trials). Encode them as statuses, not as range failures.
• You are still discovering the schema of a third-party dump. Profile for a week, then freeze rules. Premature strictness just fails every run.
• The pipeline is a one-off laptop pull. A len(df) print is enough. Do not build GX for a Thursday afternoon.

FAQ

Fail the job or quarantine the row? Fail on contract breaks: missing file, volume band, freshness, broken key, schema. Quarantine messy-but-useful rows (bad amount, duplicate ticket id) and load the rest. Do not quarantine “Monday never arrived.”

Where do the checks run? After transform, before load. Warehouse tests (dbt) are a second layer for other writers. Validating after append is how users see the bad number first.

Why did volume flap every Tuesday? You used an exact count. Use a band from a week of good runs. Holiday Tuesdays are not incidents. Zero weekday rows are.

Can I assert instead of raising DataQualityError? No. python -O strips asserts. The load will proceed. Raise.

When do I graduate from checks.py? When the same rules are copied across repos, or analysts need pass/fail history. Then dbt tests, GX, or Soda — one of them, not a secret framework in the extractor.

Should I sample 1% and call the batch clean? No. Samples miss the one null customer_id in the last partition. Check the slice you are about to load.

Production checklist

• Checks run after transform, before load, on every scheduled run — not “when we remember.”
• Grain: unique + not-null on the business key.
• Volume band and freshness lag are documented with why those numbers exist.
• Contract failures fail the job. Row-level mess has a quarantine path and an owner.
• Alerts go to a human rotation. Snoozing is a bug ticket.
• Warehouse tests exist for the same grain if Python is not the only writer.
• Quarantine paths are partitioned by date and are not infinite growth without lifecycle.

Quality is not a tool purchase. It is a gate you are willing to fail the pipeline for. Start with the four checks, make one of them page you, and only then argue about frameworks.