Design a bulk import service that validates customer spreadsheets and explains every rejected row.

“I’ll design this backward from the error report, because the person uploading is an analyst who has to fix their spreadsheet, not read a stack trace.”

Clarify. I’d ask what the rows are (I’ll assume product records for a customer’s catalog), whether a partial import is safe, and the largest file. If a partial import could leave related records inconsistent, all or nothing becomes the default.

Flow. Upload goes to object storage unchanged, and an import_job row records the file hash, the uploader, the column mapping and the status. A worker parses, validates and writes a report; the user reviews it; commit is a second, explicit step.

Parsing. CSV is parsed per RFC 4180, with encoding detected and a byte-order mark stripped. For XLSX, I read cells as text where the column is an identifier, so 00742 stays 00742, and convert date serials explicitly. Headers go through a saved mapping, so “SKU”, “Sku “ and “Item code” all map to sku for that customer. For a new customer, a language model can propose the mapping from the headers and a few sample rows; the user confirms it on screen, and it is saved, so the model never maps a column silently.

Validation. Every rule returns structured errors, never a bare exception:

from __future__ import annotations

import difflib
import re
from dataclasses import dataclass
from decimal import Decimal

@dataclass
class RowError:
    row: int          # as the spreadsheet shows it: header is row 1
    column: str
    value: str
    rule: str         # "required", "decimal", "unknown_category", "duplicate_in_file"
    message: str      # "Price must be a number, like 12.50, with no currency sign."

@dataclass
class Context:        # built once per job
    locale: str       # from the customer's saved mapping, e.g. "en_US" or "de_DE"
    categories: set[str]

    def closest_category(self, value: str) -> str | None:
        match = difflib.get_close_matches(value, sorted(self.categories), n=1, cutoff=0.6)
        return match[0] if match else None

DECIMAL_COMMA = {"de_DE", "fr_FR", "es_ES", "it_IT", "nl_NL", "pt_BR"}
POINT = re.compile(r"-?(\d{1,3}(,\d{3})+|\d+)(\.\d+)?")   # 1,234.50 or 1234.50
COMMA = re.compile(r"-?(\d{1,3}(\.\d{3})+|\d+)(,\d+)?")   # 1.234,50 or 1234,50

def parse_decimal(text: str, locale: str) -> Decimal | None:
    comma = locale in DECIMAL_COMMA
    if not (COMMA if comma else POINT).fullmatch(text):
        return None       # "$12", "NaN", "1e5", and "12.50" in a German file
    if comma:
        return Decimal(text.replace(".", "").replace(",", "."))
    return Decimal(text.replace(",", ""))

def validate_row(n: int, row: dict[str, str], ctx: Context) -> list[RowError]:
    errors: list[RowError] = []
    sku = row.get("sku", "").strip()
    if not sku:
        errors.append(RowError(n, "sku", "", "required", "SKU is required."))
    price = row.get("price", "").strip()
    if not price:
        errors.append(RowError(n, "price", "", "required", "Price is required."))
    elif parse_decimal(price, ctx.locale) is None:
        example = "12,50" if ctx.locale in DECIMAL_COMMA else "12.50"
        errors.append(RowError(n, "price", price, "decimal",
                               f"Price must be a number, like {example}, with no currency sign."))
    category = row.get("category", "").strip()
    if category and category not in ctx.categories:
        guess = ctx.closest_category(category)
        hint = f" Did you mean '{guess}'?" if guess else " See the Categories tab in the template."
        errors.append(RowError(n, "category", category, "unknown_category",
                               f"Category '{category}' not found.{hint}"))
    return errors

One bad row, and what the analyst reads:

ctx = Context(locale="en_US", categories={"Shoes", "Bags", "Hats"})
for e in validate_row(2, {"sku": "", "price": "$12", "category": "Shoez"}, ctx):
    print(e.row, e.column, e.rule, e.message)

2 sku required SKU is required.
2 price decimal Price must be a number, like 12.50, with no currency sign.
2 category unknown_category Category 'Shoez' not found. Did you mean 'Shoes'?

A row collects all its errors, not just the first, so one fix pass is enough. Cross-row checks (the same SKU twice in the file) run after the row pass and report both row numbers.

Partial acceptance. The dry run shows accepted and rejected counts. The customer picks “import valid rows” or “import nothing until all rows pass”; I default to the second when rows depend on each other. Commits are upserts keyed on (customer_id, sku), so re-uploading a corrected file of just the rejected rows is safe. An update writes only the columns present in the file, and a blank cell means “leave as is”, not “clear it”; clearing a field takes an explicit value such as #clear. For updated rows, the dry run shows field-level changes, for example price 12.50 to 13.00.

Report. On screen: errors grouped by rule, with a count and the first examples of each, because thousands of identical errors are one problem. As a download: the original rows that failed, in their original columns, plus an errors column, so the analyst fixes that file and uploads it again. The file echoes untrusted values, and a spreadsheet runs a cell that starts with =, +, - or @ as a formula. OWASP’s page on CSV injection warns that the usual fix, a leading ', stops working once Excel saves and reopens the file, which is exactly what this file is for, so the download puts a tab inside the quotes before any such cell, as that page advises for Excel. The validator trims every cell, so the tab is gone again on re-upload.

Large files. At millions of rows I stop holding the file in memory. The worker streams it and COPYs it into a staging table keyed by job_id (every check filters on it, and the rows are dropped after commit), then runs the checks as SQL: a left join against categories finds unknown values in one query, and a group by sku having count(*) > 1 finds duplicates. The report stores every error but the page shows the first 100 per rule. Progress is on the job row. Batches of 5000 are for “import valid rows”. For all or nothing, the rows already sit in the staging table, so the commit is one insert ... select ... on conflict do update in a single transaction. Either way, the commit re-runs the reference checks, because a category can be deleted between the dry run and the click, and the job moves validated → committing → committed exactly once, so a double click cannot import twice. The guard is one conditional update: UPDATE import_job SET status = 'committing' WHERE id = $1 AND status = 'validated' RETURNING id. Zero rows back means another click, on any app server, already started the commit.

On a data platform the same principle has a name. Databricks Auto Loader keeps values that don’t fit the schema in a rescued data column, with the file each one came from, instead of dropping them: keep what didn’t fit, say where it came from, and let the load go on.

First slice. CSV only, one customer’s column mapping, the row checks, the dry-run report and the download with the errors column, committing as “import valid rows”. XLSX, saved mappings for other customers and the SQL path for large files come after. To know it works, I watch how many jobs end committed within a day of the first upload, re-uploads per job, and support tickets tagged “import”. The rule that rejects the most rows is the next template fix.