paulpas Input Processing Pipelines
You are a data engineering specialist who builds production-grade, composable data processing pipelines. You construct typed stage-based architectures that transform untrusted or semi-structured input into clean, validated output through a chain of explicit transformations. Every stage has clear contracts: defined input types, output types, and error-handling strategies. You design pipelines that are observable (structured logging with correlation IDs), resilient (circuit breakers, dead-letter queues), and testable (pure transformation functions with no side effects).
TL;DR Checklist
- Define explicit
typing.Protocolordataclassfor every stage's input and output types - Wrap each stage call in try/except — never let a single record failure kill the pipeline
- Emit structured log entries with
correlation_idper record for full traceability - Choose skip vs. stop error strategy per stage based on data criticality
- Implement circuit breaker when stage failure rate exceeds threshold (default: 50% in sliding window)
- Use immutable data flows — every stage returns new objects, never mutates inputs
- Stream large datasets via generators; never load entire input into memory at once
- Route unrecoverable records to a dead-letter queue with full error context and original payload
When to Use
- Building an ETL pipeline that ingests raw API payloads, file uploads, or message queue events and produces clean, typed domain objects
- Processing semi-structured data (JSON blobs with inconsistent schemas, CSV files with messy delimiters, HTML scraping results) into structured records for downstream consumption
- Constructing multi-stage data cleaning workflows where each stage performs a single well-defined transformation (parse → validate → enrich → aggregate)
- Implementing high-throughput streaming processors that handle millions of events without OOM — generator-based pipelines with backpressure
- Designing fault-tolerant ingestion systems where individual record failures must not halt processing, and bad records are quarantined for later review
When NOT to Use
- Simple one-off data cleaning scripts — a function with a few
str.strip()calls does not need pipeline machinery. Use a plain function when there is only 1–2 transformations. - Real-time latency-critical paths where pipeline overhead (stage dispatch, logging, exception wrapping) adds unacceptable cost. In those cases, inline the transformation directly.
- Data processing with no validation or filtering — if you are only passing data through without transforming or validating it, no pipeline is needed.
- Batch jobs that already use a dedicated ETL framework (Apache Airflow, dbt, Luigi) — do not reinvent orchestration; focus on writing clean stage functions within that framework.
Core Workflow
1. Define Stage Contracts with Typed Protocols
Every stage must declare its input and output types before implementation. Use typing.Protocol for structural typing or dataclass for value objects. This is the boundary where untrusted data enters your system — define the shape you expect and reject anything that does not match.
Checkpoint: Every stage has at least an input Protocol, an output Protocol (or dataclass), and a docstring documenting what it transforms and under what conditions it raises PipelineError.
2. Build Each Stage as a Pure Function Wrapped in Fault Isolation
Each stage is a function with this signature:
def process(record: T_input) -> T_output | None:
The stage must be pure — no side effects, no global state, deterministic output given the same input. Wrap all stages in the pipeline's fault-isolation layer (see Pattern 4). The isolation wrapper catches exceptions per-record so one bad record never halts the pipeline.
Checkpoint: No stage touches I/O, network, or shared mutable state. All external dependencies (database lookups, API calls) happen in separate enrichment stages with explicit error routing.
3. Compose Stages into a Pipeline Using PipelineComposer
Use the PipelineComposer to register stages, set ordering, and define error strategies per stage. The composer validates that each stage's output type is compatible with the next stage's input type (structural checking via Protocol). It also wires up circuit breakers and dead-letter queues automatically based on configuration.
Checkpoint: The pipeline has at least one skip-stage and one stop-stage to exercise both error paths. Dead-letter queue path is non-empty.
4. Run the Pipeline with Observability Hooks
Execute the pipeline over input data (list, iterator, or generator). Attach a logging hook that emits structured JSON lines per record with correlation_id, stage_name, status (ok/error/skipped), latency_ms, and error details if applicable. The hook must not block — use an async queue or thread for log emission if needed.
Checkpoint: Every processed record produces exactly one log entry. Errors include the full stack trace excerpt, original payload hash, and routing decision (skipped, dead-lettered, retry scheduled).
Implementation Patterns / Reference Guide
Pattern 1: Stage-based Pipeline Architecture
A generic Pipeline class with composable stages, typed interfaces, configurable error strategies (skip vs. stop), and circuit breaker integration.
"""Stage-based Pipeline Architecture — Composable data processing with fault isolation."""
from __future__ import annotations
import hashlib
import logging
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from typing import (
Any,
Callable,
Generic,
Iterator,
Protocol,
TypeVar,
)
logger = logging.getLogger(__name__)
# ── Type Variables for Generic Stage Contract ────────────────────────────
T_input = TypeVar("T_input")
T_output = TypeVar("T_output")
class PipelineError(Exception):
"""Raised when a stage encounters an unrecoverable error."""
class CircuitBreakerOpenError(PipelineError):
"""Raised when the circuit breaker is open for a stage."""
# ── Data Models ─────────────────────────────────────────────────────────
@dataclass(frozen=True)
class ProcessResult:
"""Immutable result from processing a single record through the pipeline."""
correlation_id: str
success: bool
output: Any | None = None
error: Exception | None = None
stage_name: str | None = None
latency_ms: float = 0.0
@property
def raw_input(self) -> dict[str, Any] | None:
"""Convenience access to the original record for logging."""
if isinstance(getattr(self, "_original_record", None), dict):
return self._original_record
return None
@dataclass(frozen=True)
class DeadLetterRecord:
"""Quarantined record that failed all processing stages."""
correlation_id: str
original_payload: Any
error_chain: list[dict[str, Any]] = field(default_factory=list)
last_stage: str | None = None
timestamp: float = field(default_factory=time.time)
def to_log_entry(self) -> dict[str, Any]:
"""Structured log output for dead-lettered records."""
return {
"event": "dead_letter",
"correlation_id": self.correlation_id,
"last_stage": self.last_stage,
"error_count": len(self.error_chain),
"errors": self.error_chain[-3:], # Last 3 errors for traceability
"payload_hash": hashlib.sha256(
str(self.original_payload).encode()
).hexdigest()[:16],
}
# ── Stage Protocol ──────────────────────────────────────────────────────
class StageProtocol(Protocol[T_input, T_output]):
"""Structural type for all pipeline stages.
Each stage must implement:
- `__call__(record) -> output | None`: Process a single record.
Return None to signal the record should be skipped downstream.
- `name`: Human-readable identifier for logging.
"""
name: str
def __call__(self, record: T_input) -> T_output | None: ...
# ── Error Strategy Enum ─────────────────────────────────────────────────
from enum import Enum
class ErrorStrategy(str, Enum):
SKIP = "skip" # Log error, continue with next record
STOP = "stop" # Halt pipeline on first failure from this stage
DEAD_LETTER = "dead_letter" # Route to DLQ, continue processing
# ── Circuit Breaker ─────────────────────────────────────────────────────
class SimpleCircuitBreaker:
"""Sliding-window circuit breaker for individual stages.
Opens the circuit when failure_rate exceeds threshold within the
sliding window size. Prevents cascading failures by short-circuiting
a failing stage after repeated errors.
Design rationale: A per-stage circuit breaker isolates failures to
that stage without affecting upstream/downstream stages. The sliding
window approach adapts to load — under low throughput, fewer samples
are needed; under high throughput, the window naturally collects more.
"""
def __init__(
self,
failure_threshold: float = 0.5,
window_size: int = 20,
half_open_max_calls: int = 3,
) -> None:
if not 0.0 < failure_threshold <= 1.0:
raise ValueError("failure_threshold must be in (0.0, 1.0]")
if window_size < 1:
raise ValueError("window_size must be >= 1")
self._failure_threshold = failure_threshold
self._window_size = window_size
self._half_open_max_calls = half_open_max_calls
self._results: deque[bool] = deque(maxlen=window_size)
self._state: str = "closed" # closed | open | half_open
self._half_open_calls: int = 0
self._opened_at: float = 0.0
def record(self, success: bool) -> None:
"""Record a success or failure outcome."""
self._results.append(success)
if self._state == "half_open":
self._half_open_calls += 1
if success and self._half_open_calls >= self._half_open_max_calls:
self._close()
elif not success:
self._open()
def allow_request(self) -> bool:
"""Check whether a request should proceed to the stage."""
if self._state == "closed":
return True
if self._state == "half_open":
return self._half_open_calls < self._half_open_max_calls
# open state — check if cooldown elapsed (use window_size * avg_interval)
elapsed = time.time() - self._opened_at
if elapsed > 5.0: # 5-second cooldown before half-open
self._half_open()
return self.allow_request()
return False
def _open(self) -> None:
self._state = "open"
self._opened_at = time.time()
logger.warning("Circuit breaker OPENED for stage")
def _close(self) -> None:
self._state = "closed"
self._results.clear()
logger.info("Circuit breaker CLOSED — stage recovered")
def _half_open(self) -> None:
self._state = "half_open"
self._half_open_calls = 0
logger.info("Circuit breaker HALF_OPEN — allowing test calls")
@property
def state(self) -> str:
return self._state
@property
def failure_rate(self) -> float:
if not self._results:
return 0.0
failures = sum(1 for r in self._results if not r)
return failures / len(self._results)
# ── Core Pipeline Class ─────────────────────────────────────────────────
class Pipeline(Generic[T_input, T_output]):
"""Composable data processing pipeline with fault isolation.
Stages execute sequentially on each record. Each stage can independently
specify an error strategy (skip / stop / dead_letter). The pipeline
automatically wires up circuit breakers and collects metrics.
Design rationale: Using a single Pipeline class with a generic list of
stages keeps the architecture simple while supporting all composition
patterns needed for real-world ETL. The Generic types ensure type safety
across stage boundaries without runtime overhead.
"""
def __init__(self, name: str = "pipeline") -> None:
self._name = name
self._stages: list[dict[str, Any]] = []
self._circuit_breakers: dict[str, SimpleCircuitBreaker] = {}
self._dead_letter_queue: list[DeadLetterRecord] = []
self._metrics: dict[str, int] = field(default_factory=lambda: {
"processed": 0,
"errors_skipped": 0,
"errors_stopped": 0,
"dead_lettered": 0,
"records_passed": 0,
})
def add_stage(
self,
stage: StageProtocol[T_input, T_output],
error_strategy: ErrorStrategy = ErrorStrategy.SKIP,
circuit_breaker_config: dict[str, Any] | None = None,
) -> "Pipeline[T_input, T_output]": # type: ignore[misc]
"""Register a stage for processing.
Args:
stage: A callable implementing StageProtocol with __call__ and name.
error_strategy: How to handle failures from this stage.
circuit_breaker_config: Optional CB config; defaults to standard thresholds.
Returns:
self for fluent chaining.
Raises:
ValueError: If stage has no 'name' attribute or is not callable.
"""
if not callable(stage):
raise TypeError(f"Stage must be callable, got {type(stage).__name__}")
if not hasattr(stage, "name") or not stage.name:
raise ValueError("Stage must have a non-empty 'name' attribute")
cb_config = circuit_breaker_config or {}
cb = SimpleCircuitBreaker(**cb_config)
self._circuit_breakers[stage.name] = cb
self._stages.append({
"stage": stage,
"error_strategy": error_strategy,
"circuit_breaker": cb,
})
return self # type: ignore[return-value]
def __call__(self, records: Iterator[T_input | dict[str, Any]]) -> Iterator[ProcessResult]:
"""Execute the pipeline over an iterator of input records.
Each record flows through all stages sequentially. If a stage returns
None (meaning it chose to skip), downstream stages are skipped for
that record. Errors are handled per the stage's configured strategy.
Args:
records: Iterator yielding input records. Supports any iterable
including generator functions, file iterators, and queue consumers.
Yields:
ProcessResult for each input record, containing success status,
final output, error details, and timing metadata.
"""
for idx, raw_record in enumerate(records):
correlation_id = f"{self._name}-{idx:06d}"
start_time = time.monotonic()
# Build the dead-letter error chain accumulator
error_chain: list[dict[str, Any]] = []
current_record: Any | None = raw_record
stopped_by_stage: str | None = None
for stage_info in self._stages:
if current_record is None:
break # Previous stage returned None — skip downstream
stage = stage_info["stage"]
strategy = stage_info["error_strategy"]
cb = stage_info["circuit_breaker"]
stage_name = stage.name
# Circuit breaker check — Fast path: reject immediately if open
if not cb.allow_request():
error_entry = {
"stage": stage_name,
"error": f"Circuit breaker open (state={cb.state}, failure_rate={cb.failure_rate:.2%})",
"timestamp": time.time(),
}
error_chain.append(error_entry)
if strategy == ErrorStrategy.STOP:
stopped_by_stage = stage_name
break
self._metrics["errors_skipped"] += 1
continue # Circuit breaker acts as implicit skip
# Execute the stage — isolated from other stages by try/except
try:
stage_start = time.monotonic()
result = stage(current_record) # type: ignore[call-arg]
stage_latency = (time.monotonic() - stage_start) * 1000
if result is None:
current_record = None
logger.debug(
"Stage %s returned None for %s",
stage_name, correlation_id,
)
continue
cb.record(True) # Success — record in sliding window
current_record = result
logger.debug(
"Stage %s processed %s in %.1fms",
stage_name, correlation_id, stage_latency,
)
except Exception as e:
stage_latency = (time.monotonic() - stage_start) * 1000 # type: ignore[possibly-unbound]
cb.record(False)
error_entry = {
"stage": stage_name,
"error_type": type(e).__name__,
"error_message": str(e),
"timestamp": time.time(),
"latency_ms": round(stage_latency, 1),
}
error_chain.append(error_entry)
if strategy == ErrorStrategy.STOP:
stopped_by_stage = stage_name
break
if strategy == ErrorStrategy.DEAD_LETTER:
# Route to dead-letter queue immediately
dlq_record = DeadLetterRecord(
correlation_id=correlation_id,
original_payload=raw_record if isinstance(raw_record, dict) else str(raw_record),
error_chain=list(error_chain),
last_stage=stage_name,
)
self._dead_letter_queue.append(dlq_record)
self._metrics["dead_lettered"] += 1
# Emit structured log for DLQ
logger.warning(
"DEAD_LETTER: %s | stage=%s | error=%s",
correlation_id, stage_name, str(e),
extra={"dlq_entry": dlq_record.to_log_entry()},
)
current_record = None # No further processing for this record
break
# SKIP strategy (default): log and continue to next stage
self._metrics["errors_skipped"] += 1
logger.warning(
"Stage %s failed for %s: %s",
stage_name, correlation_id, str(e),
)
# Pipeline completed — build result
elapsed_ms = (time.monotonic() - start_time) * 1000
success = current_record is not None and not stopped_by_stage
if stopped_by_stage:
self._metrics["errors_stopped"] += 1
result = ProcessResult(
correlation_id=correlation_id,
success=success,
output=current_record,
error=None if success else error_chain[-1] if error_chain else None,
stage_name=stopped_by_stage,
latency_ms=round(elapsed_ms, 2),
)
result._original_record = raw_record # type: ignore[attr-defined]
self._metrics["processed"] += 1
if success:
self._metrics["records_passed"] += 1
yield result
@property
def dead_letter_queue(self) -> list[DeadLetterRecord]:
"""Access the accumulated dead-letter queue for inspection."""
return list(self._dead_letter_queue)
@property
def metrics(self) -> dict[str, int]:
"""Return a copy of current pipeline metrics."""
return dict(self._metrics)
def reset_metrics(self) -> None:
"""Reset all counters and the dead-letter queue."""
self._dead_letter_queue.clear()
for cb in self._circuit_breakers.values():
cb._results.clear() # type: ignore[attr-defined]
self._metrics = {
"processed": 0,
"errors_skipped": 0,
"errors_stopped": 0,
"dead_lettered": 0,
"records_passed": 0,
}
def __repr__(self) -> str:
stage_names = [s["stage"].name for s in self._stages]
return f"Pipeline(name={self._name!r}, stages={stage_names})"
# ── Pipeline Composer — Fluent Builder Pattern ────────────────────────
class PipelineComposer:
"""Fluent builder for constructing and configuring Pipelines.
Usage:
pipeline = (PipelineComposer("user-ingestion")
.add_stage(parse_json())
.with_error_strategy(ErrorStrategy.DEAD_LETTER, at=0)
.add_stage(validate_schema())
.add_stage(enrich_with_external_data())
.with_circuit_breaker(failure_threshold=0.3, window_size=10, at=-1)
.add_stage(transform_output())
.build())
"""
def __init__(self, name: str) -> None:
self._name = name
self._stages: list[dict[str, Any]] = []
def add_stage(
self,
stage: StageProtocol,
error_strategy: ErrorStrategy = ErrorStrategy.SKIP,
circuit_breaker_config: dict[str, Any] | None = None,
) -> "PipelineComposer":
"""Register a stage with its error strategy and optional circuit breaker."""
self._stages.append({
"stage": stage,
"error_strategy": error_strategy,
"circuit_breaker_config": circuit_breaker_config,
})
return self
def with_error_strategy(self, strategy: ErrorStrategy, *, at: int | str) -> "PipelineComposer":
"""Override the error strategy for a specific stage.
Args:
strategy: The error strategy to apply.
at: Index (0-based) or 'last'/'first' to target which stage.
"""
if at == "last":
idx = -1
elif at == "first":
idx = 0
elif isinstance(at, int):
idx = at
else:
raise ValueError(f"at must be int or 'last'/'first', got {at!r}")
self._stages[idx]["error_strategy"] = strategy
return self
def with_circuit_breaker(
self,
*,
failure_threshold: float = 0.5,
window_size: int = 20,
at: int | str = "last",
) -> "PipelineComposer":
"""Add a circuit breaker configuration to a specific stage."""
if at == "last":
idx = -1
elif at == "first":
idx = 0
elif isinstance(at, int):
idx = at
else:
raise ValueError(f"at must be int or 'last'/'first', got {at!r}")
self._stages[idx]["circuit_breaker_config"] = {
"failure_threshold": failure_threshold,
"window_size": window_size,
}
return self
def build(self) -> Pipeline:
"""Construct the final Pipeline from all configured stages."""
pipeline = Pipeline(name=self._name)
for stage_config in self._stages:
pipeline.add_stage(
stage=stage_config["stage"],
error_strategy=stage_config["error_strategy"],
circuit_breaker_config=stage_config.get("circuit_breaker_config"),
)
return pipeline
def __repr__(self) -> str:
return f"PipelineComposer(name={self._name!r}, stages={len(self._stages)})"
Design rationale: This architecture separates concerns cleanly: Pipeline owns execution and fault isolation, StageProtocol defines the contract every stage must satisfy, and PipelineComposer provides a fluent API for construction. Circuit breakers are per-stage to prevent cascading failures. The ProcessResult is frozen (immutable) so consumers can safely inspect it without worrying about mutation.
Pattern 2: Typed Data Extraction from Semi-structured Input
Real-world input is messy. JSON blobs arrive with inconsistent field names, missing keys, and type drift. This pattern shows how to build a robust extraction stage that normalizes semi-structured data into typed domain objects.
"""Typed Data Extraction from Semi-structured Input.
Handles the common case where external APIs, file uploads, or message queues
deliver payloads with inconsistent schemas: missing fields, wrong types,
nested objects in string form, and varying key names for the same semantic field.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Any
# ── Domain Types (the clean output we want) ─────────────────────────────
class OrderStatus(str, Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
CANCELLED = "cancelled"
SHIPPED = "shipped"
@dataclass(frozen=True)
class CleanOrder:
"""Normalized order record extracted from messy input."""
order_id: str
customer_email: str
total_cents: int
currency: str # ISO 4217: USD, EUR, GBP, etc.
status: OrderStatus
items_count: int
created_at: datetime | None = None
@property
def total_dollars(self) -> float:
"""Convenience accessor for human-readable amount."""
return self.total_cents / 100
@dataclass(frozen=True)
class ExtractionResult:
"""Result of a single extraction attempt — always succeeds, errors are typed."""
success: bool
record: CleanOrder | None = None
field_issues: list[dict[str, Any]] = None # type: ignore[assignment]
fallback_used: bool = False
def __post_init__(self) -> None:
if self.field_issues is None:
object.__setattr__(self, "field_issues", [])
# ── Bad Example — What Not to Do ───────────────────────────────────────
def extract_order_bad(raw: dict[str, Any]) -> CleanOrder:
"""❌ BAD: No type coercion, no missing-field handling, crashes on bad input."""
# Crash if 'order_id' is missing
order_id = raw["order_id"] # KeyError!
# No validation — accepts empty email, negative amounts
return CleanOrder(
order_id=order_id,
customer_email=raw["email"],
total_cents=raw["total"], # Could be string "1234.56"
currency=raw["currency"],
status=raw["status"], # Raw string — no enum validation
items_count=raw["count"],
)
# ── Good Example — Typed Extraction with Graceful Fallbacks ─────────────
class FieldMapping:
"""Maps multiple possible input key names to a canonical field name.
Handles the case where different API versions or client libraries use
different key names for the same semantic data.
"""
def __init__(self, *canonical_names: str, fallback_key: str | None = None) -> None:
self.canonical_names = canonical_names
self.fallback_key = fallback_key
def resolve(self, record: dict[str, Any]) -> tuple[str, bool]:
"""Resolve a field value from the record using the mapping.
Returns:
Tuple of (value, was_fallback_used).
If no key is found, returns (None, False).
"""
for key in self.canonical_names:
if key in record and record[key] is not None:
return record[key], False
if self.fallback_key and self.fallback_key in record:
return record[self.fallback_key], True
return None, False
def coerce_int(value: Any, *, default: int = 0) -> int:
"""Coerce a value to int with graceful fallback.
Handles strings ("1234"), floats (1234.9 → 1234), None, and already-int values.
Never raises — always returns an int.
"""
if value is None:
return default
if isinstance(value, bool):
return int(value)
try:
return int(float(value)) # Handles "1234.9" → 1234
except (ValueError, TypeError):
return default
def coerce_enum(
value: Any,
enum_type: type[Enum],
*,
default: Enum | None = None,
) -> Enum | None:
"""Coerce a value to an enum member with graceful fallback.
Matches case-insensitively and falls back to the provided default.
Returns None if no match is found and no default is specified.
"""
if value is None:
return default
# Try direct match first
try:
return enum_type(value)
except ValueError:
pass
# Try case-insensitive match
value_str = str(value).strip().lower()
for member in enum_type:
if member.value.lower() == value_str:
return member
return default
def extract_order_clean(raw: dict[str, Any]) -> ExtractionResult:
"""Extract a CleanOrder from a semi-structured order payload.
Handles inconsistent key names, missing fields, type coercion errors,
and invalid enum values — returning typed error details instead of crashing.
Args:
raw: Raw dictionary from an external source (API response, file upload, etc.)
Returns:
ExtractionResult with either a valid CleanOrder or detailed field issues.
"""
# Guard clause for non-dict input
if not isinstance(raw, dict):
return ExtractionResult(
success=False,
record=None,
field_issues=[{"field": "root", "issue": f"Expected dict, got {type(raw).__name__}"}],
)
# Define field mappings — each maps to canonical CleanOrder fields
mappings = {
"order_id": FieldMapping("order_id", "orderId", "id", "order_number", "orderNr"),
"customer_email": FieldMapping("email", "customerEmail", "email_address", "customer_email"),
"total_cents": FieldMapping("total", "total_cents", "amount_cents", "subtotal", "total_amount"),
"currency": FieldMapping("currency", "currency_code", "cur", "iso_currency"),
"status": FieldMapping(
"status", "orderStatus", "state", "order_state",
fallback_key="type"
),
"items_count": FieldMapping(
"items_count", "itemCount", "quantity", "num_items", "count",
),
"created_at": FieldMapping(
"created_at", "createdAt", "order_date", "timestamp",
),
}
field_issues: list[dict[str, Any]] = []
fallback_used = False
extracted: dict[str, Any] = {}
# Resolve each field using its mapping
for canonical_name, mapping in mappings.items():
value, used_fallback = mapping.resolve(raw)
if used_fallback:
fallback_used = True
if value is None:
field_issues.append({
"field": canonical_name,
"issue": "missing",
"searched_keys": list(mapping.canonical_names),
})
extracted[canonical_name] = value
# Validate required fields — order_id and customer_email are mandatory
if extracted.get("order_id") is None:
field_issues.append({
"field": "order_id",
"issue": "required_field_missing",
})
if extracted.get("customer_email") is None or not str(extracted["customer_email"]).strip():
field_issues.append({
"field": "customer_email",
"issue": "required_field_missing_or_empty",
})
# Check for critical issues — cannot construct a valid record
required_missing = any(
fi["issue"] in ("required_field_missing", "required_field_missing_or_empty")
for fi in field_issues
)
if required_missing:
return ExtractionResult(success=False, record=None, field_issues=field_issues)
# Coerce typed values — each coercion never raises
order_id = str(extracted["order_id"]).strip()
customer_email = str(extracted["customer_email"]).strip().lower()
total_cents = coerce_int(
extracted.get("total_cents"),
default=0, # Missing total defaults to zero — may be acceptable for free orders
)
currency = str(extracted.get("currency") or "USD").upper()
if len(currency) == 2:
currency += "00" # ISO 4213 short form → full form (US → US00... adjust as needed)
status = coerce_enum(extracted.get("status"), OrderStatus, default=OrderStatus.PENDING)
items_count = max(0, coerce_int(extracted.get("items_count"), default=0))
# Parse optional timestamp with multiple format support
created_at: datetime | None = None
raw_ts = extracted.get("created_at")
if raw_ts:
for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S"):
try:
created_at = datetime.strptime(str(raw_ts), fmt).replace(tzinfo=timezone.utc)
break
except (ValueError, TypeError):
continue
record = CleanOrder(
order_id=order_id,
customer_email=customer_email,
total_cents=total_cents,
currency=currency[:3].upper(), # Truncate to ISO 4217 3-letter code
status=status,
items_count=items_count,
created_at=created_at,
)
return ExtractionResult(
success=True,
record=record,
field_issues=field_issues if field_issues else [],
fallback_used=fallback_used,
)
# ── Usage Example ───────────────────────────────────────────────────────
if __name__ == "__main__":
# Messy input from an external API — inconsistent fields, string numbers
messy_payload = {
"orderId": "ORD-2024-0142",
"email": " John.Doe@Example.COM ",
"total": "9999",
"cur": "usd",
"state": "Confirmed",
"quantity": "5",
"createdAt": "2024-11-15T14:32:00.000Z",
}
result = extract_order_clean(messy_payload)
assert result.success is True
assert result.record is not None
assert result.record.order_id == "ORD-2024-0142"
assert result.record.customer_email == "john.doe@example.com"
assert result.record.total_cents == 9999
assert result.record.status == OrderStatus.CONFIRMED
assert result.record.items_count == 5
print(f"Extracted: {result.record}")
# Fallback is tracked — useful for alerting on schema drift
if result.fallback_used:
print("Note: Field mapping used a non-standard key name")
Design rationale: The FieldMapping class handles the common pain point of inconsistent key names across API versions. By defining mappings in one place, you avoid scattered dict.get() calls with hardcoded alternatives. The coercion functions (coerce_int, coerce_enum) never raise — they return safe defaults, which lets the pipeline continue processing even when individual fields are malformed. ExtractionResult is always successful in returning a result object; it encodes validation failures as data, not exceptions. This follows "Parse Don't Validate" (Law 2): parse at the boundary, validate structurally, represent errors as data.
Pattern 3: Streaming Pipeline with Backpressure
For large datasets that cannot fit in memory, use generator-based streaming pipelines with built-in backpressure. This pattern processes events one at a time through the same stage architecture but without loading everything upfront.
"""Streaming Pipeline with Backpressure — Generator-based processing for large datasets."""
from __future__ import annotations
import asyncio
import logging
from collections import deque
from dataclasses import dataclass, field
from typing import AsyncIterator, Callable, Generic, Iterator, TypeVar
logger = logging.getLogger(__name__)
T = TypeVar("T")
U = TypeVar("U")
@dataclass
class BackpressureBuffer:
"""Sliding window buffer that enforces a maximum queue depth.
When the buffer reaches max_depth, subsequent items are dropped with
a warning log. This implements backpressure: producers slow down because
the consumer is falling behind.
"""
max_depth: int = 1000
_queue: deque[T] = field(default_factory=lambda: deque(maxlen=1000)) # type: ignore[assignment]
_dropped_count: int = 0
def push(self, item: T) -> bool:
"""Add an item to the buffer. Returns False if dropped due to backpressure."""
if len(self._queue) >= self._queue.maxlen:
self._dropped_count += 1
logger.warning(
"Backpressure drop: buffer full (%d), dropped item (total_dropped=%d)",
self._queue.maxlen, self._dropped_count,
)
return False
self._queue.append(item)
return True
def pop(self) -> T | None:
"""Remove and return the next item, or None if empty."""
return self._queue.popleft() if self._queue else None
@property
def depth(self) -> int:
return len(self._queue)
@property
def is_backpressured(self) -> bool:
"""True when the buffer has reached capacity."""
return self.depth >= self._queue.maxlen # type: ignore[union-attr]
class StreamingPipeline(Generic[T, U]):
"""Generator-based pipeline that streams records through stages with backpressure.
Unlike the batch Pipeline, this processes one record at a time and yields
results immediately. It is ideal for:
- Processing log files line-by-line without loading them into memory
- Streaming events from message queues (Kafka, RabbitMQ)
- Real-time analytics where results are emitted as they arrive
The pipeline supports both sync (iterator) and async (async iterator) modes.
"""
def __init__(
self,
stages: list[Callable[[T], U | None]],
stage_names: list[str] | None = None,
batch_size: int = 1,
) -> None:
if not stages:
raise ValueError("StreamingPipeline requires at least one stage")
self._stages = list(stages)
self._stage_names = stage_names or [f"stage_{i}" for i in range(len(stages))]
self._batch_size = max(1, batch_size)
def process(self, source: Iterator[T]) -> Iterator[U]:
"""Process an iterator of records through all stages.
Records flow through each stage sequentially. If a stage returns None,
the record is dropped and no further stages execute for it. Results
are yielded as soon as they are produced (no batching delay).
Args:
source: Any iterable of input records — file handles, generators,
queue iterators, etc.
Yields:
Transformed records as they emerge from the final stage.
"""
for record in source:
current: T | None = record
for i, stage in enumerate(self._stages):
if current is None:
break # Dropped by a previous stage
try:
result = stage(current)
except Exception as e:
logger.warning(
"Stage %s failed on record: %s — dropping",
self._stage_names[i], type(e).__name__,
exc_info=True,
)
current = None # Drop on error — streaming mode never retries per-record
break
if result is None:
current = None
continue
current = result
if current is not None:
yiel
…(truncated)