Python

Verified against ChatGPT · 2026-07-17

Build a data validation pipeline that quarantines bad records instead of crashing on them

A prompt for a validation layer on an ingestion pipeline that separates "reject the whole batch" from "quarantine the bad row and keep going," backed by one named schema instead of scattered ad hoc checks.

ChatGPT (GPT-5.1)Claude (Sonnet 4.6)Claude CodeGitHub Copilot Chat

The prompt

Ready to copy — highlighted parts are example details you can swap.

Build a validation layer for the data pipeline below. The goal is a pipeline that tells you exactly which records are bad and why — not one that either crashes on the first bad row or silently accepts everything.

DATA SOURCE
A nightly batch CSV upload of ~50k customer records, processed as one DataFrame.

VALIDATION RULES
email must be a valid format; signup_date can't be in the future; country must be a valid ISO code

FAILURE POLICY
quarantine invalid rows and continue processing the rest, but abort the whole batch if over 10% of rows fail

REQUIREMENTS
1. Define the expected schema explicitly — a Pydantic model if records are processed one at a time, or a pandera DataFrameSchema if this is a bulk DataFrame pipeline. Don't validate ad hoc with scattered if checks; one schema is the single source of truth for what "valid" means.
2. Validate every record or row against the schema and classify each as valid, or invalid with the specific rule or rules it broke — not just "invalid." A record failing three rules should report all three, not just the first one that failed.
3. Apply quarantine invalid rows and continue processing the rest, but abort the whole batch if over 10% of rows fail precisely: if it says quarantine bad records and continue, invalid rows go to a separate output (a rejects table or file with the original data plus the reason) and valid rows proceed — the pipeline does not fail the whole batch for a few bad rows. If it says fail the batch above a threshold, implement that threshold explicitly rather than an arbitrary cutoff you invented.
4. Log a run summary: total records, valid count, invalid count broken down by which rule failed most often — this is what tells a human whether a new failure pattern just appeared upstream.
5. Never let a validation failure raise an unhandled exception that kills the whole pipeline run unless quarantine invalid rows and continue processing the rest, but abort the whole batch if over 10% of rows fail explicitly says that should happen for that condition.

OUTPUT FORMAT
1. The schema definition.
2. The validation and routing function (valid versus quarantined).
3. The run summary logger.
4. One worked example: a batch with a mix of valid and invalid records, showing what gets quarantined and why.
Customize the highlighted detailsoptional — the prompt above already works

Why this works

Making "quarantine and continue" versus "fail the batch above a threshold" an explicit, named decision operationalizes a choice that's usually left implicit in a bare "add validation" request, and defaults to whichever is easiest to write — which in practice means crashing on the first bad row. Requiring one schema, named specifically to the record-shape (Pydantic for one-at-a-time, pandera for a DataFrame), as the single source of truth prevents validation logic from drifting across scattered ad hoc if-checks added at different times by different people, each with a slightly different idea of what "valid" means. Requiring every broken rule per record, not just the first, matters operationally: someone fixing an upstream data issue needs the full list of what a record violated, not whichever check happened to run first and short-circuit the rest. The rule-by-rule summary logging is what actually surfaces a new upstream failure pattern — a spike in one specific rule failing is a signal worth acting on, where a bare "142 rows rejected" count is not actionable at all.

What you get back

class CustomerRecord(BaseModel): email: EmailStr signup_date: date country: str = Field(pattern=r"^[A-Z]{2}$") @field_validator("signup_date") @classmethod def not_in_future(cls, v: date) -> date: if v > date.today(): raise ValueError("signup_date is in the future") return v Run summary: 50,000 records — 48,910 valid, 1,090 invalid (612 invalid country code, 401 malformed email, 77 future signup_date). Batch proceeds: 1,090/50,000 = 2.2%, below the 10% abort threshold. Quarantined example: {"email": "not-an-email", "country": "USA"} -> rejected for both "malformed email" and "country must be 2-letter ISO code," written to rejects.csv with both reasons attached.

Verified against

ChatGPT GPT-5.1 · 2026-07-17

Claude Sonnet 4.6 · 2026-07-18

Changelog

  • 2026-07-18 Initial publish, verified against ChatGPT (GPT-5.1) and Claude (Sonnet 4.6) using pandera 0.20 and Pydantic 2.9.

Need this built into your business?

If a prompt isn't enough — custom software, built and maintained for you — that's Scult's day job.

EXPLORE CUSTOM SOFTWARE
All Python prompts

Check your AI visibility

One URL in, a 0–100 score and the exact fixes out.

RUN THE CHECK

Browse all the tools

15 tools across six categories
13 of them never send your data anywhere

Free · No signup · No trial clock

SEE THE DIRECTORY