Python Data Pipelines: Stream Input, Validate Rows and Publish Safely

Keep input size separate from correctness

A fictional supplier sends a CSV containing order_id, quantity and unit_price. Begin with a handful of known rows and a control total. Define the treatment of blank fields, repeated identifiers and malformed numbers. Increasing the file size does not make those rules less important. This guide does not claim that a particular script processed ten million rows or met a measured throughput target.

Process rows deliberately

Use a CSV parser that understands quoted delimiters instead of splitting lines manually on commas. Iterate over rows so the application need not collect the entire file merely to inspect it. Streaming input is not a guarantee of bounded memory if later code appends every row to a list. Keep the schema, accepted-row count and exception count explicit.

import csv
from decimal import Decimal

with open("orders.csv", newline="", encoding="utf-8") as source:
    for row in csv.DictReader(source):
        quantity = int(row["quantity"])
        unit_price = Decimal(row["unit_price"])
        # Validate identifiers, ranges and finite amounts before staging.
        line_amount = quantity * unit_price

Handle errors without losing the batch

The snippet illustrates parsing and is not a complete production importer. A full workflow needs row-level exception handling, a stable key and a durable staging destination. Preserve the source row number and reason for a rejected value. Decide whether any rejected record blocks the batch. Do not let an exception halfway through a file leave a reporting table that looks like a complete successful load.

Publish and replay safely

Validate staged row counts, unique keys and totals before releasing a version to consumers. Use a batch identifier and a repeatable write strategy so a replay cannot duplicate orders. Test a failure immediately before publication and another after commit but before confirmation. Keep external notifications separate from data preparation so a retried import does not resend them.

Measure the actual bottleneck

Observe parsing, transformation and destination-write time separately before adding concurrency. Choose a concurrency strategy appropriate to the actual Python runtime, libraries and workload; do not assume all modern Python builds share identical thread behavior. Record memory use and configuration during a representative test. Publish a scale result only when its data, method and limits are available.

Primary references

• Python CSV parsing
• Python decimal arithmetic