Python for Data Engineers: Projects, Packaging, Types, and Tests

In plain terms

Python connects many data systems through libraries for files, APIs, and tables. A database or query engine can perform the heavy computation while Python prepares inputs, coordinates steps, and checks results.

A notebook makes exploration convenient, but a scheduled job also needs a reproducible environment, an explicit entry point, configuration, logs, and tests. The aim is to turn an experiment into code that someone else can install, run, and diagnose.

The Python blocks build a small orders pipeline in one session; they require Python 3.12, pandas, NumPy, and PyArrow. Shell blocks run separately. The layout shows where definitions belong when moved into a project; demonstration calls stay outside package imports. Basic loops, functions, and dictionaries are assumed. The packaging section provides a separate minimal installable example.

Environments and the lock file

A dependency upgrade can change parsing defaults or result types. Pinning versions makes this variable visible and reviewable; it does not prove that the transformation is correct.

Three practices reduce this source of variation:

• One isolated environment per project. A virtual environment is a private folder of installed packages belonging to one project. Installing into the system Python, shared by everything on the machine, is how one project’s upgrade breaks another.
• Declared, pinned dependencies. The declaration (pyproject.toml) says what the project needs; the lock file records the exact version of every package, including the ones pulled in indirectly. Both are committed, and the server installs from the lock file, so it gets the same libraries the tests ran against.
• A pinned Python version. 3.12 and 3.9 differ in syntax and standard library. Write the version into the project.

uv manages environments and project dependencies. A venv/pip workflow can also be reproducible if all dependencies and the interpreter are pinned; a loose requirements file alone is insufficient. The explicit --package option below avoids relying on changing project-generation defaults. 3.12 selects a Python series, not an exact patch release.

uv init --package --python 3.12 orders-pipeline
cd orders-pipeline
uv python pin 3.12
uv add pandas pyarrow numpy
uv add --dev pytest ruff mypy
uv sync --locked                  # refuses a stale lock
uv run --locked orders-pipeline   # generated console entry, initially a greeting

uv sync --locked checks that declarations and the lock agree and refuses a stale lock. Plain uv sync can update it; --frozen skips the freshness check. Review and commit both files. Record the Python patch, OS, and native libraries too: the lock can select platform-specific artifacts and does not make different machines identical. See uv locking and syncing.

A project layout that can be deployed

orders-pipeline/
├── pyproject.toml          # name, version, dependencies, tool configuration
├── uv.lock                 # exact versions, committed
├── .python-version
├── README.md
├── src/
│   └── orders_pipeline/
│       ├── __init__.py
│       ├── __main__.py     # `python -m orders_pipeline` starts here
│       ├── config.py       # settings from the environment
│       ├── extract.py      # talks to sources
│       ├── transform.py    # pure functions: data in, data out
│       ├── load.py         # writes to destinations
│       └── models.py       # record shapes (dataclasses / pydantic)
└── tests/
    ├── conftest.py         # shared fixtures
    ├── test_transform.py
    └── data/
        └── orders_sample.csv   # 50 rows, committed

Two decisions in that tree carry most of the value.

Code lives under src/ as an importable package. Installing it avoids relying on the repository root being on the import path. During development uv normally installs it in editable mode, so edits remain visible. A separate wheel-install test checks the actual distribution; the directory layout alone does not guarantee that all needed files are included.

Extract, transform, and load are separate modules, and transform is pure. Pure means the functions take data and return data without touching files, databases, or the network. That makes them testable in milliseconds with a fifty-row sample. The I/O is pushed to the edges, where it is thin and can be replaced by a fake in tests. This structure comes up again in the testing section and in the pipeline-patterns article (5).

From source tree to an installed wheel

An import package (orders_pipeline) is the code Python imports. A distribution (orders-pipeline) is the installable project described by pyproject.toml. Its build backend makes a wheel, an archive of installable code and metadata. The wheel does not contain the complete environment; runtime dependencies still need installation. requires-python declares compatibility, while the interpreter pin selects a runtime. Keep package imports free of demonstration I/O.

The generated console command calls the function named in [project.scripts]. Running python -m orders_pipeline additionally needs __main__.py; the larger tree above is a target layout, not a promise that uv init creates every file. The separate exercise below packages a tiny same-currency sum so installation can be checked without copying the entire pipeline. It does not enforce the later order-record money contract.

# Separate, disposable packaging exercise (Bash).
set -euo pipefail
package_lab="$(mktemp -d)"
cd "$package_lab"
mkdir -p src/orders_pipeline tests
cat > pyproject.toml <<'TOML'
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "orders-pipeline"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = []

[project.scripts]
orders-pipeline = "orders_pipeline:main"
TOML
cat > src/orders_pipeline/__init__.py <<'PYTHON'
import argparse
from decimal import Decimal


def total_amount(amounts: list[Decimal]) -> Decimal:
    if any(not value.is_finite() for value in amounts):
        raise ValueError("amounts must be finite")
    return sum(amounts, Decimal(0))


def main() -> int:
    parser = argparse.ArgumentParser(description="Sum same-currency amounts.")
    parser.add_argument("amounts", nargs="*", type=Decimal)
    args = parser.parse_args()
    print(total_amount(args.amounts))
    return 0
PYTHON
cat > src/orders_pipeline/__main__.py <<'PYTHON'
from orders_pipeline import main

raise SystemExit(main())
PYTHON
cat > tests/test_amount.py <<'PYTHON'
from decimal import Decimal

import pytest

from orders_pipeline import total_amount


def test_exact_total():
    assert total_amount([Decimal("19.99"), Decimal("5.50")]) == Decimal("25.49")


def test_empty_total():
    assert total_amount([]) == Decimal(0)


def test_nonfinite_rejected():
    with pytest.raises(ValueError):
        total_amount([Decimal("NaN")])
PYTHON
uv python pin 3.12
uv add --dev pytest mypy ruff
uv sync --locked
uv run --locked ruff check .
uv run --locked ruff format --check .
uv run --locked mypy --strict src/
uv run --locked pytest -q
uv run --locked python -m orders_pipeline 19.99 5.50
# 25.49
uv build --wheel
uv venv "$package_lab/wheel-env" --python 3.12
uv pip install --python "$package_lab/wheel-env/bin/python" dist/*.whl
cd "$(mktemp -d)"                  # run outside the source checkout
"$package_lab/wheel-env/bin/python" -m orders_pipeline 19.99 5.50
# 25.49
"$package_lab/wheel-env/bin/orders-pipeline" 19.99 5.50
# 25.49

The development tests exercise an editable installation. Building the wheel and invoking both entry points from a different directory checks that the artifact contains importable code and working entry points. It is a smoke test, not the full pipeline integration test. Build dependencies have their own resolution; pin the backend or use a controlled build environment when reproducing the build artifact is required. See uv building distributions and pytest test layout guidance.

Configuration and secrets

Anything that differs between a laptop, the test environment, and production (hostnames, bucket names, dates, credentials) is configuration and does not live in code. Read it from environment variables through a typed object, so that a missing or malformed setting fails at startup rather than three hours in:

import os
from collections.abc import Mapping
from dataclasses import dataclass, field

@dataclass(frozen=True)
class Settings:
    db_host: str
    db_password: str = field(repr=False)
    raw_bucket: str
    batch_size: int = 5000

    @classmethod
    def from_env(cls, env: Mapping[str, str] | None = None) -> "Settings":
        values = os.environ if env is None else env
        required = ("DB_HOST", "DB_PASSWORD", "RAW_BUCKET")
        missing = [k for k in required if not values.get(k, "").strip()]
        if missing:
            raise ValueError(f"missing settings: {', '.join(missing)}")
        try:
            size = int(values.get("BATCH_SIZE", "5000"))
        except ValueError:
            raise ValueError("BATCH_SIZE must be a positive integer") from None
        if size <= 0:
            raise ValueError("BATCH_SIZE must be a positive integer")
        return cls(values["DB_HOST"], values["DB_PASSWORD"], values["RAW_BUCKET"], size)

try:
    Settings.from_env({})
except ValueError as exc:
    print(exc)
# missing settings: DB_HOST, DB_PASSWORD, RAW_BUCKET
settings = Settings.from_env({"DB_HOST": "warehouse.internal",
                              "DB_PASSWORD": "fake-demo-value", "RAW_BUCKET": "raw-orders"})
print(settings.db_host, settings.batch_size)
# warehouse.internal 5000

The injected dictionary makes this example independent of your real environment. repr=False hides the password from the generated representation, not from memory or every serializer. Standard os.environ does not load .env files: a shell or explicitly configured loader must do that. pydantic-settings offers parsing and validation, but positive limits and secret handling still need configuration. This settings example illustrates a database job; the local file pipeline below instead accepts paths as command-line arguments.

Language features that matter for data

Generators: files larger than memory

A list comprehension builds every parsed record before returning. A generator yields records incrementally, so a consumer such as sum can process them without retaining the whole file. The following measures traced Python allocations, not the process’s full resident memory.

import json, tempfile, tracemalloc
from pathlib import Path

workdir = Path(tempfile.mkdtemp())
big = workdir / "events.jsonl"
with big.open("w") as f:
    for i in range(200_000):
        f.write(json.dumps({"id": i, "amount": i % 100, "country": "KR" if i % 3 else "US"}) + "\n")

def parse(line):
    return json.loads(line)

# loads everything, then returns
def read_all(path):
    with open(path) as f:
        return [parse(line) for line in f]

# yields one record at a time
def read_records(path):
    with open(path) as f:
        for line in f:
            yield parse(line)

def total_amount(records):
    return sum(r["amount"] for r in records)

for reader in (read_all, read_records):
    tracemalloc.start()
    total = total_amount(reader(big))
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    assert total == 9_900_000
    print(f"{reader.__name__:13} total={total} peak={peak / 1e3:.0f} KB")
# read_all      total=9900000 peak=80195 KB   # varies by machine
# read_records  total=9900000 peak=3 KB   # varies by machine

Both paths assert the total of 9,900,000. Peak allocations depend on the runtime. Streaming stays bounded here because records are small and the consumer keeps only a sum. A growing deduplication dictionary, a very large record, driver prefetching, or a downstream list() changes that conclusion. The later pipeline intentionally collects a small daily batch; it is not an unbounded streaming implementation.

Context managers: closing what was opened

A context manager runs its exit logic when control leaves the block, including ordinary exceptions. For an opened file this closes it:

try:
    with open(big) as f:
        first = f.readline()
        raise ValueError("something went wrong halfway")
except ValueError as exc:
    print(exc, "| file closed:", f.closed)
# something went wrong halfway | file closed: True

The file closed despite the exception. Check each library’s contract: a SQLite connection context commits or rolls back a transaction but does not close the connection. Use close() or contextlib.closing where required. No Python cleanup mechanism is guaranteed to run after a forced process kill. See SQLite connection contexts.

Records with a declared shape

With a dictionary, d[key] raises KeyError for a missing key; d.get(key) returns a default, often None. Neither declares a complete record contract. Here the dataclass holds already typed values, while parse_order validates raw CSV text at the boundary.

from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation

KNOWN_CURRENCIES = {"USD", "KRW", "EUR", "JPY"}

@dataclass(frozen=True)
class Order:
    order_id: str
    customer_id: str
    amount: Decimal
    currency: str
    ordered_at: datetime

def parse_order(raw: Mapping[str, object]) -> Order:
    def text(name: str) -> str:
        value = raw.get(name)
        if not isinstance(value, str) or not value.strip():
            raise ValueError(f"{name}: non-empty text required")
        return value.strip()

    order_id, customer_id = text("order_id"), text("customer_id")
    try:
        amount = Decimal(text("amount"))
    except InvalidOperation:
        raise ValueError("amount: invalid decimal") from None
    if not amount.is_finite() or abs(amount) >= Decimal("1e16"):
        raise ValueError("amount: finite and abs(value) < 1e16 required")
    if amount != amount.quantize(Decimal("0.01")):
        raise ValueError("amount: at most two decimal places required")
    currency = text("currency")
    if currency not in KNOWN_CURRENCIES:
        raise ValueError("currency: unsupported code")
    try:
        stamp = datetime.fromisoformat(text("ordered_at").replace("Z", "+00:00"))
    except ValueError:
        raise ValueError("ordered_at: invalid ISO timestamp") from None
    if stamp.utcoffset() is None:
        raise ValueError("ordered_at: must be timezone-aware")
    return Order(order_id, customer_id, amount, currency, stamp.astimezone(timezone.utc))

good = dict(order_id="A1", customer_id="C9", amount="19.99", currency="USD",
            ordered_at="2026-03-14T09:00:00Z")
ok = parse_order(good)
print(ok.amount, ok.ordered_at.isoformat())
# 19.99 2026-03-14T09:00:00+00:00
for bad in (dict(amount="abc"), dict(ordered_at="2026-03-14T09:00:00"), dict(currency="usd$")):
    try:
        parse_order({**good, **bad})
    except ValueError as exc:
        print("rejected:", exc)
# rejected: amount: invalid decimal
# rejected: ordered_at: must be timezone-aware
# rejected: currency: unsupported code

The parser requires IDs, finite decimal text, a supported currency, and an aware timestamp, then normalizes time to UTC. The four currencies and two-decimal storage rule are this example’s contract, not a complete currency standard; real currency scales and refund rules must be chosen explicitly. The bound fits decimal128(18, 2) used below. Dataclasses do not enforce annotations at runtime, and frozen=True prevents ordinary field reassignment rather than validating inputs. Pydantic can parse fields, but business rules such as allowed currencies still need validators or constraints. See dataclasses.

Type hints

The two transforms the rest of the article uses, with their signatures spelled out:

def daily_revenue(orders: list[Order], day: date) -> dict[str, Decimal]:
    """Sum amount per currency for orders placed on `day` (UTC)."""
    totals: dict[str, Decimal] = {}
    for o in orders:
        if o.ordered_at.astimezone(timezone.utc).date() == day:
            totals[o.currency] = totals.get(o.currency, Decimal("0")) + o.amount
    return totals

def deduplicate(orders: list[Order]) -> list[Order]:
    """Collapse identical records; do not guess which conflicting record wins."""
    unique: dict[str, Order] = {}
    for order in orders:
        if order.order_id in unique and unique[order.order_id] != order:
            raise ValueError(f"conflicting order_id: {order.order_id}")
        unique[order.order_id] = order
    return [unique[key] for key in sorted(unique)]

sample = [ok,
          parse_order({**good, "order_id": "A2", "amount": "5.50",
                       "ordered_at": "2026-03-14T23:59:59Z"}),
          parse_order({**good, "order_id": "A3", "amount": "12000", "currency": "KRW",
                       "ordered_at": "2026-03-15T00:00:01Z"})]
print(daily_revenue(sample, date(2026, 3, 14)))
# {'USD': Decimal('25.49')}
print(len(deduplicate(sample + [ok])))
# 3

Type hints describe permitted calls to readers and static checkers. Mypy can detect daily_revenue(sample, "2026-03-14") as a wrong argument type; it cannot inspect tomorrow’s CSV. Runtime validation handles that boundary. Duplicate IDs collapse only when all fields agree; conflicts fail because order time does not identify the newest revision. Resolving revisions requires a separate, trustworthy version field and tie policy.

Dates, decimals, and paths

from decimal import ROUND_HALF_UP
from zoneinfo import ZoneInfo

# aware, UTC; not datetime.now()
now = datetime.now(timezone.utc)
print(now.tzinfo, datetime(2026, 3, 14).tzinfo)
# UTC None
seoul = datetime(2026, 3, 14, 0, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Asia/Seoul"))
print(seoul.isoformat())
# 2026-03-14T09:00:00+09:00
price = Decimal("19.99") * 3
print(price, price.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), Decimal(19.99))
# 59.97 59.97 19.989999999999998436805981327779591083526611328125
p = Path("/raw/orders") / "date=2026-03-14" / "part-0.parquet"
print(p.parent, p.suffix, p.name)
# /raw/orders/date=2026-03-14 .parquet part-0.parquet

Three habits. datetime objects are timezone-aware and UTC internally; the second value on the first line, None, is a naive datetime and a bug waiting to happen. Money is a Decimal constructed from a string, because constructing from a float carries the float’s error in, as the long third value shows. Paths are Path objects rather than strings joined with +, so the parent, suffix, and name are available without string surgery.

Logging rather than print

import logging, sys

# Application setup; a library should not install handlers on import.
log = logging.getLogger("orders_pipeline_demo")
log.handlers.clear()
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(levelname)s %(name)s %(message)s"))
log.addHandler(handler)
log.setLevel(logging.INFO)
log.propagate = False
log.info("loading orders date=%s rows=%d", "2026-03-14", 3)
log.warning("rejected rows=%d; publication decision pending", 2)
# INFO orders_pipeline_demo loading orders date=2026-03-14 rows=3
# WARNING orders_pipeline_demo rejected rows=2; publication decision pending

Logging adds levels, handlers, and configurable formatting. Add timestamps, run IDs, durations, and row counts in application configuration; do not log credentials or entire source records. Inside an exception handler, log.exception includes the traceback. Logging an error does not make the job fail: re-raise or return a nonzero status when processing cannot continue. The demonstration handler uses stdout for readable output; the deployment may route logs to stderr or a collector.

Exceptions with information

class SourceMissing(Exception):
    """Raised when an expected input file or partition does not exist."""

landing = workdir / "landing"
landing.mkdir()
(landing / "orders_20260314.csv").write_text(
    "order_id,customer_id,amount,currency,ordered_at\n"
    "A1,C9,19.99,USD,2026-03-14T09:00:00Z\n"
    "A2,C3,5.50,USD,2026-03-14T23:59:59Z\n"
    "A3,C4,abc,KRW,2026-03-14T12:00:00Z\n")

def extract(day: date, source_dir: Path) -> Path:
    p = source_dir / f"orders_{day:%Y%m%d}.csv"
    if not p.is_file():
        raise SourceMissing(f"expected {p.name}; did the vendor deliver?")
    return p

print(extract(date(2026, 3, 14), landing).name)
# orders_20260314.csv
try:
    extract(date(2026, 3, 15), landing)
except SourceMissing as exc:
    print("SourceMissing:", exc)
# SourceMissing: expected orders_20260315.csv; did the vendor deliver?

Two rules. except Exception: pass converts every failure into silent wrong output, which the earlier prerequisite articles describe as the outcome to avoid most. And when catching something, either handle it fully or re-raise it with context; a named exception with the filename in its message turns an investigation into a one-line diagnosis.

Libraries in daily use

The pandas memory note

pd.read_csv without chunking materializes the table. Memory depends on types, parsing, temporary arrays, and library versions as well as file size. Lab 2 measures each complete process rather than assuming a fixed file-to-RAM multiplier.

Consider SQL close to the source, chunked pandas, or batch-oriented Arrow processing. Streaming support depends on the operation: a global sort or high-cardinality aggregation can still retain substantial state. Compare memory, semantics, and elapsed time on representative inputs before selecting an engine.

SQL from Python: parameters

The standard library’s sqlite3 is enough to show the rule, and every other driver follows it with its own placeholder (? here, %s in psycopg):

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customers (customer_id TEXT, name TEXT)")
conn.executemany("INSERT INTO customers VALUES (?, ?)", [("C1", "Kim"), ("C2", "O'Brien")])

name = "O'Brien"
# wrong: the value is pasted into the SQL text
try:
    conn.execute(f"SELECT customer_id FROM customers WHERE name = '{name}'").fetchall()
except sqlite3.OperationalError as exc:
    print("f-string:", exc)
# f-string: near "Brien": syntax error
print("parameter:", conn.execute("SELECT customer_id FROM customers WHERE name = ?", (name,)).fetchall())
# parameter: [('C2',)]

hostile = "x' OR '1'='1"
print(conn.execute("SELECT count(*) FROM customers WHERE name = ?", (hostile,)).fetchone())
# (0,)
print(conn.execute(f"SELECT count(*) FROM customers WHERE name = '{hostile}'").fetchone())
# (2,)
conn.close()

Only run this deliberately broken query against the disposable in-memory database. Bound parameters keep values separate from SQL syntax, including apostrophes and the hostile string. They do not parameterize table or column names; use a driver’s identifier composition or an allowlist for those. Transaction and connection cleanup are separate concerns.

Tests

Pure transforms accept data and return results without a database. Tests can therefore compare small, explicit inputs with expected outputs. Pytest discovers test_ functions in test_*.py files. The calls below are a notebook-friendly demonstration, not a replacement for pytest discovery, fixtures, or reporting. In a project, import the definitions and put only tests in the test module.

def make_order(**changes: str) -> Order:
    return parse_order({**good, **changes})

def test_daily_revenue_sums_per_currency():
    orders = [make_order(amount="10.00"), make_order(order_id="A2", amount="5.50"),
              make_order(order_id="A3", currency="KRW", amount="12000")]
    assert daily_revenue(orders, date(2026, 3, 14)) == {"USD": Decimal("15.50"), "KRW": Decimal("12000")}

def test_daily_revenue_excludes_other_days():
    assert daily_revenue([make_order(ordered_at="2026-03-15T00:00:01Z")], date(2026, 3, 14)) == {}

def test_deduplicate_keeps_one_per_order_id():
    for copies in (1, 2, 5):
        assert deduplicate([make_order()] * copies) == [make_order()]

def test_duplicate_conflict_is_rejected():
    try:
        deduplicate([make_order(amount="10.00"), make_order(amount="12.00")])
    except ValueError as exc:
        assert "conflicting order_id" in str(exc)
    else:
        raise AssertionError("conflicting versions were accepted")

for test in (test_daily_revenue_sums_per_currency, test_daily_revenue_excludes_other_days,
             test_deduplicate_keeps_one_per_order_id, test_duplicate_conflict_is_rejected):
    test()
print("4 passed")
# 4 passed

These tests check currency separation, the UTC day, exact duplicates, and conflicting IDs. Assertions must check business outcomes, not just whether a function returns. They are small enough for routine development checks; later tests cover input parsing and file publication.

Testing the edges without a real database

Use temporary directories and recorded responses for deterministic I/O tests. A local SQLite test does not prove warehouse-specific SQL, transaction, or authentication behavior; add integration tests against the target system for those contracts. Pytest supplies tmp_path automatically. The direct calls below supply temporary paths by hand.

import pyarrow as pa
import pyarrow.parquet as pq

ORDER_SCHEMA = pa.schema([
    pa.field("order_id", pa.string(), nullable=False),
    pa.field("customer_id", pa.string(), nullable=False),
    pa.field("amount", pa.decimal128(18, 2), nullable=False),
    pa.field("currency", pa.string(), nullable=False),
    pa.field("ordered_at", pa.timestamp("us", tz="UTC"), nullable=False),
])

def load(orders: list[Order], out: Path, day: date) -> Path:
    if any(o.ordered_at.astimezone(timezone.utc).date() != day for o in orders):
        raise ValueError("order outside target UTC day")
    rows = [dict(order_id=o.order_id, customer_id=o.customer_id, amount=o.amount,
                 currency=o.currency, ordered_at=o.ordered_at)
            for o in deduplicate(orders)]
    table = pa.Table.from_pylist(rows, schema=ORDER_SCHEMA)
    dest = out / f"date={day.isoformat()}" / "part-0.parquet"
    dest.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(dir=dest.parent, suffix=".tmp", delete=False) as f:
        pending = Path(f.name)
    try:
        pq.write_table(table, pending)
        os.replace(pending, dest)
    finally:
        pending.unlink(missing_ok=True)
    return dest

def test_load_round_trip_and_rerun(tmp_path):
    orders = [make_order(order_id="A2"), make_order(order_id="A1")]
    dest = load(orders, tmp_path, date(2026, 3, 14))
    first = pq.ParquetFile(dest).read()
    load(list(reversed(orders)), tmp_path, date(2026, 3, 14))
    second = pq.ParquetFile(dest).read()
    assert first.equals(second)
    assert first.schema == ORDER_SCHEMA
    assert first.column("order_id").to_pylist() == ["A1", "A2"]
    assert first.column("amount").to_pylist() == [Decimal("19.99"), Decimal("19.99")]
    assert len(list(dest.parent.glob("*.parquet"))) == 1

def test_empty_output_keeps_schema(tmp_path):
    dest = load([], tmp_path, date(2026, 3, 14))
    table = pq.ParquetFile(dest).read()
    assert table.num_rows == 0 and table.schema == ORDER_SCHEMA

for test in (test_load_round_trip_and_rerun, test_empty_output_keeps_schema):
    with tempfile.TemporaryDirectory() as folder:
        test(Path(folder))
print("2 passed")
# 2 passed

The schema remains explicit even for an empty day. Sorting by ID stabilizes row order; rereading verifies values and types, not just row count. A temporary file in the destination directory is replaced only after Parquet writing succeeds. This gives single-file atomic replacement on a supporting local filesystem, not a multi-file transaction, object-store commit, concurrent-writer lock, or power-loss durability guarantee. Repeated runs deliberately add diagnostic files when rejects occur; idempotency here refers to the published daily data.

Tooling

uv run --locked ruff check .
uv run --locked ruff format --check .
uv run --locked mypy src/
uv run --locked pytest -q
# Development fixes, after inspecting the diff:
# uv run ruff check . --fix
# uv run ruff format .

Run all four before each commit; a pre-commit hook makes it automatic. The formatter matters more than it sounds: in a codebase where every file looks the same, a diff shows only the change that matters, and review time goes to logic rather than spacing.

The entry point

The entry point receives the date and both directory paths explicitly. It rejects an unexpected header, quarantines malformed records, and blocks publication by default when any record is rejected. --allow-rejects is an explicit partial-publication policy for this demonstration, not a universally acceptable failure threshold.

import argparse, csv
from typing import Any

FIELDS = ["order_id", "customer_id", "amount", "currency", "ordered_at"]

def read_orders(path: Path) -> tuple[list[Order], list[dict[str, Any]]]:
    accepted: list[Order] = []
    rejected: list[dict[str, Any]] = []
    with path.open(newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f, strict=True)
        if reader.fieldnames != FIELDS:
            raise ValueError("unexpected CSV header")
        for row in reader:
            try:
                if None in row or any(v is None for v in row.values()):
                    raise ValueError("wrong CSV field count")
                accepted.append(parse_order(row))
            except ValueError as exc:
                rejected.append({"line_end": reader.line_num, "raw": row, "reason": str(exc)})
    return accepted, rejected

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Process one UTC day of orders.")
    parser.add_argument("day", type=date.fromisoformat)
    parser.add_argument("--input-dir", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--allow-rejects", action="store_true")
    args = parser.parse_args(argv)
    raw = extract(args.day, args.input_dir)
    orders, rejected = read_orders(raw)
    orders = deduplicate(orders)
    if any(o.ordered_at.date() != args.day for o in orders):
        raise ValueError("source contains orders outside target UTC day")
    blocked = bool(rejected) and not args.allow_rejects
    log.info("day=%s accepted=%d rejected=%d blocked=%s",
             args.day, len(orders), len(rejected), blocked)
    if args.dry_run:
        return 2 if blocked else 0
    if rejected:
        # Diagnostic evidence is written even when publication is blocked.
        args.output_dir.mkdir(parents=True, exist_ok=True)
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".jsonl",
                                         prefix=f"rejects-{args.day}-", dir=args.output_dir,
                                         delete=False) as f:
            for row in rejected:
                f.write(json.dumps({"source": str(raw), **row}) + "\n")
    if blocked:
        return 2
    load(orders, args.output_dir / "raw", args.day)
    return 0

run_args = ["2026-03-14", "--input-dir", str(landing), "--output-dir", str(workdir / "result")]
assert main(run_args + ["--dry-run"]) == 2
assert not (workdir / "result").exists()
assert main(run_args) == 2
assert not (workdir / "result" / "raw").exists()
assert main(run_args + ["--allow-rejects"]) == 0
print("dry run, blocked publication, explicit partial publication: passed")
# dry run, blocked publication, explicit partial publication: passed

Each call logs counts; dry-run performs reads and validation but creates no files. Status 2 means the reject policy blocked publication; explicit permission writes two valid rows. CSV syntax errors, schema errors, conflicting IDs, and wrong-day records fail the run rather than becoming a successful partial result. Diagnostics retain source path, physical end-line number, parsed fields, and reason. Preserve the source version or checksum separately for exact replay, and protect quarantine data like its source. A run date alone cannot reproduce mutable input. A blocked run can leave an older partition in place; downstream jobs must honor run status and input version. The application entry point should use raise SystemExit(main()) so the scheduler receives its status.

When to leave Python

Large Python-level loops can spend much of their time creating objects and dispatching operations. SQL or array-library operations may move that work into optimized native code. Some native libraries already use threads; a Python application is not necessarily single-threaded. Profile the actual bottleneck before moving work: I/O waits, small batches, and complex row validation can justify ordinary Python.

Lab

The Python solutions continue the session above; shell labs use fresh temporary projects. The million-row CSV is synthetic benchmark data, not production money accounting. Runtime and memory need measuring rather than linear extrapolation. To build the larger project, move definitions into the shown modules, add imports, and keep fixture generation and demonstration calls in tests or examples.

1. Reproduce the environment. Build a second environment from the same lock. Change a dependency declaration without updating the lock and verify that --locked refuses it. Restore the declaration; do not edit lock entries manually.

set -euo pipefail
lab_dir="$(mktemp -d)"
cd "$lab_dir"
uv init --package --python 3.12 lock-demo
cd lock-demo
uv python pin 3.12
uv lock
uv sync --locked
cp pyproject.toml pyproject.before.toml
uv add --frozen packaging
if uv sync --locked; then
    printf 'Unexpected: stale lock accepted\n'
    exit 1
else
    printf 'Expected: stale lock rejected\n'
fi
mv pyproject.before.toml pyproject.toml
uv sync --locked
UV_PROJECT_ENVIRONMENT="$lab_dir/rebuilt-env" uv sync --locked
uv run --locked lock-demo

2. Compare memory without exhausting the machine. Run whole-table pandas, a CSV iterator, and Arrow batches in separate processes. Validate currency-labelled totals before comparing peaks and elapsed time.

import subprocess, sys, textwrap, time
import numpy as np
import pandas as pd

rng = np.random.default_rng(0)
n = 1_000_000
pd.DataFrame({
    "order_id": [f"A{i}" for i in range(n)],
    "customer_id": rng.integers(1, 200_000, n),
    "amount": np.round(rng.gamma(2.0, 20.0, n), 2),
    "currency": rng.choice(["USD", "KRW", "EUR"], n),
    "ordered_at": "2026-03-14T09:00:00Z",
}).to_csv(workdir / "orders_1m.csv", index=False)
path = str(workdir / "orders_1m.csv")
print(f"{(workdir / 'orders_1m.csv').stat().st_size / 1e6:.1f} MB on disk")
# 45.2 MB on disk

def peak_rss_mb(snippet: str) -> float:
    """Run the snippet in a fresh process and report that process's peak resident memory."""
    program = textwrap.dedent(snippet) + """
import resource, sys
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(peak / 1e6 if sys.platform == "darwin" else peak * 1024 / 1e6)
"""
    if sys.platform not in ("darwin", "linux"):
        raise RuntimeError("RSS lab requires macOS or Linux")
    out = subprocess.run([sys.executable, "-c", program], capture_output=True, text=True, check=True)
    lines = out.stdout.strip().splitlines()
    if len(lines) > 1:
        result = __import__("ast").literal_eval(lines[-2])
        assert result == expected_totals
    return float(lines[-1])

expected_totals = pd.read_csv(path).groupby("currency")["amount"].sum().round(2).to_dict()

approaches = {
    "import only": f"""
        import pandas, pyarrow
    """,
    "pandas read_csv": f"""
        import pandas as pd
        df = pd.read_csv({path!r})
        print(df.groupby("currency")["amount"].sum().round(2).to_dict())
    """,
    "csv generator": f"""
        import csv
        from collections import defaultdict
        totals = defaultdict(float)
        with open({path!r}, newline="") as f:
            for row in csv.DictReader(f):
                totals[row["currency"]] += float(row["amount"])
        print({{k: round(v, 2) for k, v in sorted(totals.items())}})
    """,
    "pyarrow batches": f"""
        import pyarrow as pa, pyarrow.csv as pcsv
        from collections import defaultdict
        totals = defaultdict(float)
        with pcsv.open_csv({path!r}) as reader:
            for batch in reader:
                grouped = pa.Table.from_batches([batch]).group_by("currency").aggregate([("amount", "sum")])
                for row in grouped.to_pylist():
                    totals[row["currency"]] += row["amount_sum"]
        print({{k: round(v, 2) for k, v in sorted(totals.items())}})
    """,
}
for name, code in approaches.items():
    t = time.perf_counter()
    print(f"{name:16} peak {peak_rss_mb(code):5.0f} MB  {time.perf_counter() - t:.1f} s")
# import only      peak    95 MB  0.4 s   # varies by machine
# pandas read_csv  peak   267 MB  1.5 s   # varies by machine
# csv generator    peak     9 MB  2.0 s   # varies by machine
# pyarrow batches  peak   155 MB  1.1 s   # varies by machine

3. Validate at the boundary. Add 1,000 invalid rows to 10,000 valid ones, retain each rejected row and reason, and inspect reason counts. Decide separately whether those rejects permit publication.

from collections import Counter
import numpy as np

rows = ["order_id,customer_id,amount,currency,ordered_at"]
rng = np.random.default_rng(1)
for i in range(10_000):
    rows.append(f"B{i},C{rng.integers(1, 500)},{rng.gamma(2.0, 20.0):.2f},USD,"
                f"2026-03-14T{rng.integers(0, 24):02d}:00:00Z")
for i in range(1_000):
    kind = i % 3
    if kind == 0:
        # missing amount
        rows.append(f"X{i},C1,,USD,2026-03-14T09:00:00Z")
    elif kind == 1:
        # naive timestamp
        rows.append(f"X{i},C1,10.00,USD,2026-03-14T09:00:00")
    else:
        # amount as text
        rows.append(f"X{i},C1,N/A,USD,2026-03-14T09:00:00Z")
src = landing / "validation_sample.csv"
src.write_text("\n".join(rows) + "\n")

orders, rejected = read_orders(src)
quarantine = workdir / "validation_quarantine.jsonl"
quarantine.write_text("\n".join(json.dumps(r) for r in rejected) + "\n", encoding="utf-8")
print(len(orders), len(rejected))
# 10000 1000
print(Counter(r["reason"] for r in rejected).most_common())
# [('amount: non-empty text required', 334), ('ordered_at: must be timezone-aware', 333), ('amount: invalid decimal', 333)]
first_reject = json.loads(quarantine.read_text().splitlines()[0])
print(first_reject["raw"]["order_id"], first_reject["reason"])
# X0 amount: non-empty text required

4. Write the edge tests. The midnight boundary, an unknown currency, and an empty input day. Make one fail first, then add the transform that makes it pass.

def test_midnight_boundary():
    inside = make_order(ordered_at="2026-03-15T00:30:00+09:00")
    outside = make_order(order_id="A2", ordered_at="2026-03-15T00:00:00Z")
    assert daily_revenue([inside, outside], date(2026, 3, 14)) == {"USD": Decimal("19.99")}

def test_empty_day():
    assert daily_revenue([], date(2026, 3, 14)) == {}

def test_invalid_input_is_rejected():
    for bad in ({"currency": "XXX"}, {"amount": "NaN"}, {"amount": "Infinity"},
                {"amount": "0.001"}, {"order_id": ""}, {"ordered_at": "2026-03-14T09:00:00"}):
        try:
            make_order(**bad)
        except ValueError:
            pass
        else:
            raise AssertionError(f"accepted bad input: {bad}")

# A behavioral mutation: interpreting the source's calendar day as the UTC day.
source_stamp = datetime.fromisoformat("2026-03-15T00:30:00+09:00")
try:
    assert source_stamp.date() == date(2026, 3, 14)
except AssertionError:
    print("local-date comparison fails the UTC-day requirement")
# local-date comparison fails the UTC-day requirement
assert source_stamp.astimezone(timezone.utc).date() == date(2026, 3, 14)
for test in (test_midnight_boundary, test_empty_day, test_invalid_input_is_rejected):
    test()
print("3 passed")
# 3 passed

5. Check rerun behavior. Run twice with the same input and compare logical values, schema, and bytes in the same environment.

import hashlib

assert main(run_args + ["--allow-rejects"]) == 0
output = workdir / "result" / "raw" / "date=2026-03-14" / "part-0.parquet"
first_table = pq.ParquetFile(output).read()
first_hash = hashlib.sha256(output.read_bytes()).hexdigest()
assert main(run_args + ["--allow-rejects"]) == 0
second_table = pq.ParquetFile(output).read()
second_hash = hashlib.sha256(output.read_bytes()).hexdigest()
assert first_table.equals(second_table)
assert first_table.schema == ORDER_SCHEMA
print(first_table.num_rows, first_hash == second_hash)
# 2 True

6. Break the SQL. Insert a thousand customers including one named O'Brien, then query for that name with an f-string and with a parameter. Try a hostile value through both.

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customers (customer_id TEXT PRIMARY KEY, name TEXT)")
names = [(f"C{i}", f"Customer {i}") for i in range(1, 1000)] + [("C1000", "O'Brien")]
conn.executemany("INSERT INTO customers VALUES (?, ?)", names)

target = "O'Brien"
try:
    conn.execute(f"SELECT customer_id FROM customers WHERE name = '{target}'")
except sqlite3.OperationalError as exc:
    print("f-string:", exc)
# f-string: near "Brien": syntax error
print("parameter:", conn.execute("SELECT customer_id FROM customers WHERE name = ?", (target,)).fetchone())
# parameter: ('C1000',)

hostile = "nobody' OR '1'='1"
print("parameter:", conn.execute("SELECT count(*) FROM customers WHERE name = ?", (hostile,)).fetchone())
# parameter: (0,)
print("f-string: ", conn.execute(f"SELECT count(*) FROM customers WHERE name = '{hostile}'").fetchone())
# f-string:  (1000,)
conn.close()

7. Move one step out of Python. Take the per-currency total over the million-row file and compute it once with a Python loop over the rows and once with pyarrow’s grouped aggregation, which runs in C++. Confirm they agree and time both. If DuckDB is installed, add a third version as one SQL statement over the same file.

import time
import numpy as np
import pandas as pd
import pyarrow.csv as pcsv

csv_path = workdir / "orders_1m.csv"
if not csv_path.exists():
    rng = np.random.default_rng(0)
    n = 1_000_000
    pd.DataFrame({
        "order_id": [f"A{i}" for i in range(n)],
        "customer_id": rng.integers(1, 200_000, n),
        "amount": np.round(rng.gamma(2.0, 20.0, n), 2),
        "currency": rng.choice(["USD", "KRW", "EUR"], n),
        "ordered_at": "2026-03-14T09:00:00Z",
    }).to_csv(csv_path, index=False)

table = pcsv.read_csv(csv_path)
t = time.perf_counter()
totals = {}
for row in table.select(["currency", "amount"]).to_pylist():
    totals[row["currency"]] = totals.get(row["currency"], 0.0) + row["amount"]
loop_seconds = time.perf_counter() - t

t = time.perf_counter()
grouped = table.group_by("currency").aggregate([("amount", "sum")]).to_pylist()
arrow_seconds = time.perf_counter() - t

python_totals = {k: round(v, 2) for k, v in totals.items()}
arrow_totals = {r["currency"]: round(r["amount_sum"], 2) for r in grouped}
assert python_totals == arrow_totals
print(python_totals == arrow_totals)
# True
print(f"python loop {loop_seconds:.2f} s, arrow group_by {arrow_seconds:.3f} s")
# python loop 0.68 s, arrow group_by 0.003 s   # varies by machine