Process CSV Rows Without Loading Everything

import csv
import tempfile
from collections.abc import Iterator
from pathlib import Path
from typing import Any

REQUIRED = ("id", "email", "age")


def stream_rows(
    source_path: Path,
    rejects_path: Path,
) -> Iterator[dict[str, Any]]:
    with (
        source_path.open("r", encoding="utf-8", newline="") as source,
        rejects_path.open("w", encoding="utf-8", newline="") as rejects,
    ):
        reader = csv.DictReader(source)
        fieldnames = reader.fieldnames or []
        missing = [name for name in REQUIRED if name not in fieldnames]
        if missing:
            raise ValueError("missing columns: " + ", ".join(missing))

        writer = csv.DictWriter(rejects, fieldnames=["line", "reason"])
        writer.writeheader()

        for line_no, row in enumerate(reader, start=2):
            raw_age = (row.get("age") or "").strip()
            try:
                age = int(raw_age)
            except ValueError:
                writer.writerow({"line": line_no, "reason": "age is not an integer"})
                continue
            if not 18 <= age <= 120:
                writer.writerow({"line": line_no, "reason": "age outside 18..120"})
                continue
            email = (row.get("email") or "").strip().lower()
            if "@" not in email:
                writer.writerow({"line": line_no, "reason": "email is invalid"})
                continue
            yield {
                "id": (row.get("id") or "").strip(),
                "email": email,
                "age": age,
            }


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        source = Path(tmp) / "people.csv"
        rejects = Path(tmp) / "rejects.csv"
        source.write_text(
            "id,email,age\n"
            "1,[email protected],36\n"
            "2,[email protected],n/a\n"
            "3,[email protected],17\n",
            encoding="utf-8",
        )
        valid = list(stream_rows(source, rejects))
        print(valid)
        print(rejects.read_text(encoding="utf-8").strip())


if __name__ == "__main__":
    main()