Python for Data Engineers

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.

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 gives a project its own package installation location, normally isolated from system site-packages. It shares the host operating system and is not a security sandbox. Recreate it from declared dependencies rather than copying the folder between machines.
• Declared, pinned dependencies. The declaration (pyproject.toml) says what the project needs; the lock file records resolved dependencies, including indirect ones, with versions and selection conditions. Commit both and install the intended dependency groups from that lock. A universal lock can contain different selections for different Python versions or platforms; matching the file alone does not make two installations identical.
• 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. A pure function returns the same result for the same inputs and has no observable side effects, such as mutating its caller’s data or writing to a database. Avoid hidden inputs such as the current clock or mutable global settings; pass the run date and relevant rules explicitly. These transforms assume the stated decimal arithmetic context. Separating I/O makes small, explicit examples easier to test, while file and database behavior still needs its own tests.

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.

Development tests can use an editable installation. Building a 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, and validate the required values at startup. The annotation describes the object; the explicit checks below reject missing or malformed input:

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. The scratch directory it creates holds the 200,000-line file and later the landing and result folders, so it stays in use until the cleanup step after the entry point section.

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. A generator is a one-pass iterator: after consumption, iterate over a newly created generator to reread the file. Its body starts on iteration, so file-opening and parsing errors can occur during consumption rather than at the call to read_records.

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. In read_records, however, yield suspends execution inside the file’s with block. Breaking out of the consumer’s loop does not itself exit that block while the generator remains alive. For early termination, explicitly close this generator, for example by wrapping it in contextlib.closing, instead of relying on garbage collection. 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

This pipeline uses aware timestamps normalized to UTC. The displayed None identifies a naive datetime: it has no offset that identifies an instant. Naive values can be appropriate for a local wall-time contract, but they must not be treated as UTC implicitly. Construct decimal amounts from strings to avoid importing a float’s binary approximation. Decimal arithmetic still uses a precision and rounding context; these small-batch examples assume the default 28-digit precision. Choose sufficient precision for maximum amounts and batch totals, and an explicit rounding policy for operations such as division. 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?

Catching Exception and doing nothing suppresses a broad class of errors and can let processing continue with incomplete results. Catch the expected exception when there is a defined recovery or reject policy; otherwise let it propagate or re-raise with useful context. SourceMissing identifies an unavailable expected file, not the underlying cause: a wrong directory, delayed delivery, or access problem still needs investigation. Do not put credentials or full sensitive records into exception messages.

Libraries in daily use

Need	Library	Note
HTTP / APIs	httpx or requests	reuse a client or session (networking article); set timeouts
Tables in memory	pandas, polars	choose by data types, operations, and measured memory/time
Parquet, Arrow	pyarrow	reads and writes Parquet in batches
Local SQL on files	duckdb	query CSV/Parquet with SQL, no server; suits tests and exploration
Databases	psycopg, sqlalchemy, sqlite3	parameterized queries, never string formatting; see below
Cloud storage	boto3, google-cloud-storage, azure-storage-blob	or fsspec/s3fs for a file-like view of buckets
Validation	pydantic	at the boundary
Config files	pyyaml, tomllib	TOML is in the standard library from 3.11
Command line	argparse, typer	a job takes its run date as an argument

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. Measure the 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 .

These tools check different things: Ruff finds configured lint issues, its formatter standardizes layout, mypy checks annotated type relationships, and pytest exercises assertions. Configure hooks or CI to run the relevant checks. Consistent formatting reduces spacing-only diffs; none of these tools alone establishes that the business result is correct.

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; accepted here counts records after identical duplicates are collapsed, whereas rejected counts rejected source rows. Do not add these two numbers to infer total input rows or use them directly as a rejection-rate denominator. Dry-run performs reads and validation but creates no files; it does not test destination permissions or the write path. 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.

The session created workdir with tempfile.mkdtemp(), which is never deleted automatically. Removing it also removes the generated events file, the landing folder, the published Parquet file, and the reject file written by the demonstration run:

import shutil

shutil.rmtree(workdir)
print(workdir.exists())
# False

A tempfile.TemporaryDirectory() context removes its directory automatically, but wrapping a single earlier block in it would delete files that the later examples read. That is why the earlier blocks share one directory that is removed once, here.

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.

References

• Astral, Locking and syncing
• Astral, Building distributions
• pytest, Good Integration Practices
• Python, sqlite3 — Connection context manager
• Python, dataclasses — Data Classes