Importing a 5 GB, 30-million-row CSV with generators and batches without exhausting memory

# ======================================================================
# Q: You have to import a 5 GB CSV with 30 million rows. How do you do it
# without running out of memory?
#
# Streaming with generators: I read line by line, validate and write in batches.
# Never readlines() or loading the whole file with pandas.read_csv. For a clean
# import the fastest option is PostgreSQL COPY into a staging table and then
# INSERT ... SELECT with an upsert.
# ======================================================================

import csv
from itertools import batched           # Python 3.12+
from pydantic import BaseModel, ValidationError

class EventRow(BaseModel):
    external_id: str
    host: str
    severity: int
    occurred_at: datetime

def parse(path):
    with open(path, newline='', encoding='utf-8') as f:
        for lineno, raw in enumerate(csv.DictReader(f), start=2):
            try:
                yield EventRow(**raw).model_dump(), None
            except ValidationError as e:
                yield None, {'line': lineno, 'raw': raw, 'errors': e.errors()}

def import_file(session, path, import_id):
    for chunk in batched(parse(path), 5000):
        good = [row for row, err in chunk if row]
        bad = [err for row, err in chunk if err]
        if good:
            upsert_events(session, good)
        if bad:
            session.execute(insert(RejectedRow),
                            [{'import_id': import_id, **b} for b in bad])
        session.commit()                 # small transactions, not one huge one