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:

Volume — did roughly the expected number of rows arrive?
Uniqueness — is the business key actually unique?
Completeness — are critical columns non-null?
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.