Validate Spreadsheet Uploads with Pydantic
A customer uploads a spreadsheet of records to be processed. Reading it is easy; trusting it is not:
df = pd.read_excel(upload)
for row in df.to_dict("records"):
create_record(**row) # TypeError, ValueError, or worse: succeeds with nonsense
One row has a quantity of "twelve", another a date as 45012, a third an empty email. Some fail loudly, some are inserted as-is, and the error message a customer sees is a traceback mentioning a column name they have never heard of.
Root Cause
A spreadsheet has no schema. Excel stores a date as a serial number, a reference with leading zeros as an integer, and a column of numbers as text the moment one cell contains a note. pandas then infers a dtype per column from what it happens to see, so the same upload template produces different types depending on the data — a quantity column is int64 in one file and object in the next. Without a declared contract, the validation ends up scattered through the processing code as if statements that each handle one case, and the first row that fails aborts the run with no report of the other nineteen problems in the file.
Minimal Diagnostic
Find out what the file actually contains before writing rules for it.
# pip install pandas openpyxl
from pathlib import Path
import pandas as pd
def profile(path: Path, sheet: int | str = 0, sample: int = 3) -> None:
df = pd.read_excel(path, sheet_name=sheet, dtype=object)
print(f"{path.name}: {len(df)} row(s), {len(df.columns)} column(s)")
for column in df.columns:
values = df[column]
kinds = values.dropna().map(lambda v: type(v).__name__).value_counts().to_dict()
blanks = int(values.isna().sum())
examples = values.dropna().head(sample).tolist()
print(f" {str(column)!r:<22} types={kinds} blanks={blanks} e.g. {examples}")
duplicated = df.duplicated().sum()
print(f" fully duplicated rows: {duplicated}")
if __name__ == "__main__":
profile(Path("uploads/records.xlsx"))
records.xlsx: 214 row(s), 5 column(s)
'Reference ' types={'str': 211, 'int': 3} blanks=0 e.g. ['A-0041', 'A-0042', 'A-0043']
'Customer' types={'str': 214} blanks=0 e.g. ['Northgate Ltd', 'Bevan & Co', 'Ash Ltd']
'Quantity' types={'int': 210, 'str': 4} blanks=0 e.g. [12, 3, 40]
'Due Date' types={'Timestamp': 198, 'int': 11, 'str': 5} blanks=0 e.g. [...]
'Email' types={'str': 209} blanks=5 e.g. ['[email protected]', ...]
fully duplicated rows: 2
Mixed types in three columns, five blank emails, two duplicate rows and a header with a trailing space — all of it before a single record is processed.
Fix: One Model, Every Row Reported
Declare the contract as a model, validate every row, and collect the failures rather than stopping at the first.
# pip install pydantic pandas openpyxl
import re
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Annotated
import pandas as pd
from pydantic import BaseModel, ConfigDict, EmailStr, Field, ValidationError, field_validator
EXCEL_EPOCH = datetime(1899, 12, 30) # Excel's day 0, accounting for its 1900 bug
class Record(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
reference: Annotated[str, Field(pattern=r"^[A-Z]-\d{4}$")]
customer: Annotated[str, Field(min_length=2, max_length=120)]
quantity: Annotated[int, Field(gt=0, le=10_000)]
due_date: date
email: EmailStr | None = None
@field_validator("due_date", mode="before")
@classmethod
def parse_excel_date(cls, value):
if isinstance(value, (int, float)) and not isinstance(value, bool):
return (EXCEL_EPOCH + timedelta(days=float(value))).date() # changed: serial -> date
if isinstance(value, str):
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d %b %Y"):
try:
return datetime.strptime(value.strip(), fmt).date()
except ValueError:
continue
raise ValueError(f"unrecognised date {value!r}; use YYYY-MM-DD")
return value
@field_validator("email", mode="before")
@classmethod
def blank_to_none(cls, value):
return None if value is None or str(value).strip() == "" else value
def normalise(name: str) -> str:
return re.sub(r"[^a-z0-9]+", "_", str(name).strip().lower()).strip("_")
@dataclass
class UploadResult:
records: list[Record]
errors: list[tuple[int, str, str]] # (spreadsheet row, field, message)
def validate_upload(path: Path, sheet: int | str = 0) -> UploadResult:
df = pd.read_excel(path, sheet_name=sheet, dtype=object)
df.columns = [normalise(column) for column in df.columns]
records, errors = [], []
for offset, row in enumerate(df.to_dict("records")):
line = offset + 2 # changed: spreadsheet row, header is row 1
clean = {k: (None if pd.isna(v) else v) for k, v in row.items()}
try:
records.append(Record(**clean))
except ValidationError as error:
for detail in error.errors(): # changed: every problem in the row
field = ".".join(str(p) for p in detail["loc"]) or "(row)"
errors.append((line, field, detail["msg"]))
return UploadResult(records, errors)
if __name__ == "__main__":
result = validate_upload(Path("uploads/records.xlsx"))
print(f"{len(result.records)} valid record(s), {len(result.errors)} problem(s)")
for line, field, message in result.errors[:10]:
print(f" row {line}, {field}: {message}")
198 valid record(s), 21 problem(s)
row 14, quantity: Input should be a valid integer, unable to parse string as an integer
row 27, reference: String should match pattern '^[A-Z]-\d{4}$'
row 88, due_date: unrecognised date 'end of month'; use YYYY-MM-DD
row 91, email: value is not a valid email address
Two decisions make this report usable. Reporting the spreadsheet row number rather than the DataFrame index means a customer can open the file and go straight to the cell. And iterating error.errors() rather than printing the exception lists every problem in a row, so a row with three bad fields is fixed in one pass rather than three uploads.
extra="forbid" is worth the strictness: a file with an unexpected column is usually the wrong template, and finding that out at row one is better than processing 200 rows that are missing a field nobody noticed was required.
Variant Fix 1: Cross-Row Rules
Some rules are about the file, not the row — uniqueness, totals, a reference appearing twice:
# pip install pydantic
from collections import Counter
def cross_row_checks(result: UploadResult) -> list[tuple[int, str, str]]:
problems = []
seen: dict[str, int] = {}
for line, record in zip(range(2, 2 + len(result.records)), result.records):
if record.reference in seen:
problems.append((line, "reference",
f"duplicate of row {seen[record.reference]}"))
else:
seen[record.reference] = line
total = sum(record.quantity for record in result.records)
if total > 100_000:
problems.append((0, "(file)", f"total quantity {total:,} exceeds the 100,000 limit"))
per_customer = Counter(record.customer for record in result.records)
for customer, count in per_customer.items():
if count > 50:
problems.append((0, "(file)", f"{customer} appears {count} times; expected at most 50"))
return problems
Keeping these separate from the model is deliberate. A Pydantic model validates one object, and forcing file-level rules into it through shared state makes the model unusable anywhere else. Running them as a second pass over the validated records keeps both halves simple and means the row-level errors are already known when the file-level ones are checked.
Variant Fix 2: Reporting Back to the Uploader
The person who uploaded the file needs the problems in a form they can act on, which means their column names and a file they can open:
# pip install pandas
from pathlib import Path
import pandas as pd
def error_report(result: UploadResult, original_columns: dict[str, str], out: Path) -> Path:
rows = [{"Row": line,
"Column": original_columns.get(field, field),
"Problem": message}
for line, field, message in result.errors]
frame = pd.DataFrame(rows).sort_values(["Row", "Column"])
frame.to_excel(out, index=False)
return out
Mapping the normalised field name back to the heading the customer used — due_date to Due Date — removes the last piece of internal vocabulary from the report. A spreadsheet of problems is also the format most likely to be acted on, because it can be sorted, filtered and worked through row by row in the same tool the data came from.
All or Nothing, or Partial Acceptance?
The policy question matters more than the code. Three answers are defensible and the wrong one is silent.
Rejecting the whole file on any error is the right default for financial data and anything where a partial import is hard to reverse. The uploader fixes and resubmits, and the system's state is always the result of one complete file. Accepting the valid rows and reporting the rest suits high-volume operational data where waiting for a perfect file costs more than processing 198 of 214 records — but only if the rejected rows are tracked, or they are simply lost. Holding the whole file for human review fits low-volume, high-value uploads.
Whichever policy applies, state it in the response. An uploader who sees "214 rows received" and is not told that 16 were rejected will assume all 214 were processed, and will discover otherwise weeks later when something does not reconcile.
Verification
Test the model against the shapes real files contain, not just the happy path.
# pip install pydantic pytest
from datetime import date
import pytest
from pydantic import ValidationError
GOOD = {"reference": "A-0041", "customer": "Northgate Ltd", "quantity": 12,
"due_date": "2026-10-01", "email": "[email protected]"}
def test_accepts_a_clean_row() -> None:
record = Record(**GOOD)
assert record.due_date == date(2026, 10, 1) and record.quantity == 12
def test_converts_an_excel_serial_date() -> None:
record = Record(**{**GOOD, "due_date": 46296})
assert record.due_date == date(2026, 10, 1), f"got {record.due_date}"
def test_rejects_a_worded_quantity() -> None:
with pytest.raises(ValidationError) as caught:
Record(**{**GOOD, "quantity": "twelve"})
assert caught.value.errors()[0]["loc"] == ("quantity",)
def test_blank_email_becomes_none() -> None:
assert Record(**{**GOOD, "email": " "}).email is None
def test_unexpected_column_is_rejected() -> None:
with pytest.raises(ValidationError):
Record(**{**GOOD, "notes": "whatever"})
def test_every_error_in_a_row_is_reported() -> None:
with pytest.raises(ValidationError) as caught:
Record(**{**GOOD, "quantity": -1, "reference": "nope", "due_date": "soon"})
fields = {error["loc"][0] for error in caught.value.errors()}
assert fields == {"quantity", "reference", "due_date"}, f"only reported {fields}"
The last test is the one that protects the user experience. It fails the moment someone restructures the validation into sequential checks that stop at the first problem, which is the change that turns one upload cycle into four.
FAQ
Pydantic or pandera for spreadsheets? Pydantic for row-by-row validation with per-row error reporting; pandera for column-level checks on a DataFrame. See pandera column not in DataFrame.
How do I handle a 200,000-row upload? Validate in chunks and cap the error list — nobody reads 40,000 errors. Report the first hundred and the counts per field.
Should the model coerce or reject?
Coerce only unambiguous representations, such as an Excel date serial. Coercing "twelve" to 12 guesses at intent.
Can I generate the upload template from the model?
Yes — the field names and constraints are available through model_json_schema(), which is enough to write a headed template with a validation note per column.
Related
- Validating Document Data with Schemas — the validation workflow end to end
- Fix pandera Column Not in DataFrame Error — column-level checks on a DataFrame
- Reading Excel Files with Python — getting the file in before validating it
- Cleaning Messy CSV Data with pandas — cleaning what validation rejects