# Input Processing Pipelines > Builds composable data processing pipelines that validate, transform, filter, and aggregate structured or semi-structured input through typed stages with error handling and observability. - Skill: `paulpas/input-processing-pipelines` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/input-processing-pipelines` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/input-processing-pipelines/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/input-processing-pipelines --- # 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.Protocol` or `dataclass` for 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_id` per 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: ```python 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. ```python """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. ```python """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. ```python """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)