</>
pythonbeginner

Dataclasses as Pipeline Data Models

Use Python dataclasses to define typed, immutable data models passed between pipeline stages.

python
from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class RawEvent:
    event_id: str
    user_id:  int
    action:   str
    ts:       float

@dataclass(frozen=True)
class EnrichedEvent:
    event_id:    str
    user_id:     int
    action:      str
    created_at:  datetime
    hour:        int
    day_of_week: str

def enrich(raw: RawEvent) -> EnrichedEvent:
    dt = datetime.utcfromtimestamp(raw.ts)
    return EnrichedEvent(raw.event_id, raw.user_id, raw.action, dt, dt.hour, dt.strftime('%A'))

print(enrich(RawEvent('e1', 1, 'click', 1_700_000_000.0)))

Use Cases

  • typed pipeline stages
  • data modeling
  • functional ETL

Tags

#dataclasses#typing#pipeline#data-modeling

Related Snippets

Similar patterns you can reuse in the same workflow.

pythonintermediate

attrs Classes as Immutable Pipeline Records

Use attrs to create fast, validated, immutable record types for data pipeline stage outputs.

Best for: typed pipeline records

#attrs#data-modeling
pythonadvanced

Python ETL Pipeline Example

Complete extract-transform-load pipeline with error handling, logging, and incremental processing.

Best for: Automating data ingestion from CSV to warehouse

#etl#pipeline
pythonintermediate

Retry Logic for Data Pipelines

Configurable retry decorator with exponential backoff and jitter for resilient data pipeline tasks.

Best for: Resilient API calls in data pipelines

#retry#resilience
pythonadvanced

Databricks Notebook Data Pipeline

Databricks notebook with Delta Lake reads, transformations, merge operations, and table optimization.

Best for: Medallion architecture data pipelines on Databricks

#databricks#delta-lake