Design a streaming CSV importer with validation

01The design, in the six-step framework

1. Requirements and use cases

An operator uploads a CSV file that may be many gigabytes. The importer reads it row by row, checks each row against a configurable set of rules, writes valid rows to a destination and records invalid ones with their line number and reason. One bad row never stops the import. At the end the operator gets a report with counts and errors. A file that lacks a required column fails immediately, before any row is processed.

Out of scope: resuming an interrupted import, type coercion into domain objects, and parallel parsing. The report below assumes a single reader; the reported round was a senior machine-coding exercise, so working code beats breadth.

Constraints: memory stays constant regardless of file size, so nothing may collect rows in a list; rules are added without editing the importer; the error list itself is capped, or a file with a billion bad rows would exhaust memory through the report.

2. Core entities and relationships

• Row: an immutable record of the fields and the source line number, which every error needs.
• RowSource: anything iterable over rows. CsvSource adapts the standard csv module.
• Rule: one validation returning None or a message. Required, IsInt and Unique are examples.
• Sink: where valid rows go; ListSink serves tests.
• Importer: pulls rows through the rules into the sink. Report and RowError record the result.

The importer composes its report and aggregates an ordered list of rules supplied by the caller. Source and sink are used through protocols, so the same pipeline can read a network stream or write to a database.

3. Interfaces and key methods

@dataclass(frozen=True)
class Row:
    line: int
    fields: Mapping[str, str]

class RowSource(Protocol):
    def __iter__(self) -> Iterator[Row]: ...

class Rule(Protocol):
    def check(self, row: Row) -> str | None:
        """None if the row passes, otherwise a message."""

class Sink(Protocol):
    def write(self, row: Row) -> None: ...

class Importer:
    def __init__(self, source: RowSource, rules: Sequence[Rule], sink: Sink,
                 max_errors: int = 1000) -> None: ...
    def run(self) -> "Report":
        """Stream every row; never loads the file; bad rows go to the report."""

Invariants: every row read ends up counted in exactly one of imported and rejected; a rejected row never reaches the sink; errors never exceeds max_errors, and truncated is True whenever an error was dropped; rows are held one at a time.

4. Design patterns and the choices behind them

Iterator for streaming

The force is file size. csv.DictReader is already lazy, and exposing rows through an iterator keeps that property across the whole pipeline. The tempting rows = list(reader) gives easier code and fails on the first large file. A consequence of streaming is that a rule cannot look ahead, and stateful rules such as Unique grow with distinct values, so that memory must be stated out loud.

Rules as specification objects

Each rule object answers one question about one row. Adding "email must contain @" is one small class plus a line in the configuration, which is what open-closed asks for. A single validate function with many ifs was rejected because it cannot be configured per file or tested per rule.

Collect all failures or stop at the first

A chain of responsibility would stop at the first failing handler. Here the operator wants every problem in a row reported at once, so they fix the file in one pass; _validate runs all rules. The cost is that a later rule may see data an earlier rule already rejected, so rules must tolerate bad values, and IsInt handles None for that reason. The chain is right when rules are expensive or depend on earlier ones.

Adapter around the library

CsvSource is an adapter: it converts the library's dicts into our Row and attaches the line number. Swapping in a JSON-lines source touches nothing else. Header validation sits in its constructor so a bad file fails before work begins (see errors and validation).

5. Core flow and code

1. The caller opens the file and builds CsvSource(stream, required=["id", "age"]). The header is read and checked; a missing column raises ValueError here.
2. The caller builds Importer(source, [Required("id"), IsInt("age"), Unique("id")], sink) and calls run.
3. run pulls one Row. _validate asks each rule in order and collects the RowError records.
4. No problems: the row goes to sink.write and imported rises.
5. Problems: rejected rises and errors are added up to the cap; beyond it truncated is set and counting continues.
6. When the iterator ends, the report is returned. The whole file passed through a single row's worth of memory, apart from Unique's set and the capped error list.

from __future__ import annotations
import csv
from dataclasses import dataclass, field
from typing import Iterable, Iterator, Mapping, Protocol, Sequence, TextIO

@dataclass(frozen=True)
class Row:
    line: int
    fields: Mapping[str, str]

@dataclass(frozen=True)
class RowError:
    line: int
    rule: str
    message: str

@dataclass
class Report:
    imported: int = 0
    rejected: int = 0
    errors: list[RowError] = field(default_factory=list)
    truncated: bool = False

class RowSource(Protocol):
    def __iter__(self) -> Iterator[Row]: ...

class CsvSource:
    """Adapts the csv module to our Row type; reads lazily."""
    def __init__(self, stream: TextIO, required: Sequence[str] = ()) -> None:
        self._reader = csv.DictReader(stream)
        missing = set(required) - set(self._reader.fieldnames or [])
        if missing:
            raise ValueError(f"missing columns: {sorted(missing)}")

    def __iter__(self) -> Iterator[Row]:
        for fields in self._reader:
            yield Row(self._reader.line_num, fields)

class Rule(Protocol):
    def check(self, row: Row) -> str | None: ...

class Sink(Protocol):
    def write(self, row: Row) -> None: ...

@dataclass(frozen=True)
class Required:
    column: str
    def check(self, row: Row) -> str | None:
        return None if (row.fields.get(self.column) or "").strip() else f"{self.column} is empty"

@dataclass(frozen=True)
class IsInt:
    column: str
    def check(self, row: Row) -> str | None:
        try:
            int(row.fields.get(self.column) or "")
            return None
        except ValueError:
            return f"{self.column} is not an integer"

class Unique:
    """Stateful: remembers values seen in this run."""
    def __init__(self, column: str) -> None:
        self.column, self._seen = column, set()
    def check(self, row: Row) -> str | None:
        value = row.fields.get(self.column)
        if value in self._seen:
            return f"duplicate {self.column}: {value}"
        self._seen.add(value)
        return None

class ListSink:
    def __init__(self) -> None:
        self.rows: list[Row] = []
    def write(self, row: Row) -> None:
        self.rows.append(row)

class Importer:
    def __init__(self, source: Iterable[Row], rules: Sequence[Rule], sink: Sink,
                 max_errors: int = 1000) -> None:
        self._source, self._rules, self._sink = source, rules, sink
        self._max_errors = max_errors

    def _validate(self, row: Row) -> list[RowError]:
        found = []
        for rule in self._rules:
            message = rule.check(row)
            if message is not None:
                found.append(RowError(row.line, type(rule).__name__, message))
        return found

    def run(self) -> Report:
        report = Report()
        for row in self._source:
            problems = self._validate(row)
            if not problems:
                self._sink.write(row)
                report.imported += 1
                continue
            report.rejected += 1
            room = self._max_errors - len(report.errors)
            report.errors.extend(problems[:max(0, room)])
            report.truncated = report.truncated or len(problems) > room
        return report

6. Extensions, concurrency and what interviewers probe

• Cross-row validation: a uniqueness check on a huge file cannot hold every key; use a Bloom filter with a second pass over candidates, or push the constraint to the destination.
• Batching the sink: wrap Sink in a buffered decorator that writes every 1,000 rows; flushing at the end must be guaranteed, so add a close method.
• Parallel validation: use a bounded queue between one reader and N validators; line numbers keep errors meaningful, but output order is lost and stateful rules need a lock.
• Malformed lines: a row with too many columns appears under DictReader's rest key; a generic rule can reject it. Embedded newlines in quoted fields mean a record may span lines, which is why line_num is the reader's value and not a loop counter.
• Test first: an in-memory io.StringIO with good, bad and duplicate rows; assert counts, exact error lines, and the error cap.
• Common mistake: raising on the first invalid row, which loses the rest of the import, and accumulating an unbounded error list.

02Where this question was reported

• raw.githubusercontent.com · Build a streaming CSV importer with row validation; parse and evaluate a simple expression/DSL
• LLD Mastery · Commercial tag: JSON parser (recursive descent, tokenization)