Classes and dataclasses for pipelines

from dataclasses import dataclass
from datetime import date
from decimal import Decimal

@dataclass(frozen=True, slots=True)
class Order:
    order_id: int
    customer_id: int
    order_date: date
    status: str
    amount: Decimal

    def __post_init__(self):
        if self.amount <= 0:
            raise ValueError(f"order {self.order_id}: amount <= 0")

    @classmethod
    def from_row(cls, r: dict) -> "Order":
        return cls(int(r["order_id"]), int(r["customer_id"]),
                   date.fromisoformat(r["order_date"]),
                   r["status"].strip().lower(), Decimal(r["amount"]))

o = Order.from_row({"order_id": "101", "customer_id": "1",
                    "order_date": "2025-01-10",
                    "status": " Delivered", "amount": "1198.00"})
print(o)
print(o == Order(101, 1, date(2025, 1, 10), "delivered",
                 Decimal("1198.00")))
try:
    Order(116, 8, date(2025, 5, 2), "delivered", Decimal("-50"))
except ValueError as e:
    print("rejected:", e)

Order(order_id=101, customer_id=1, order_date=datetime.date(2025, 1, 10), status='delivered', amount=Decimal('1198.00'))
True
rejected: order 116: amount <= 0

import csv, json
from abc import ABC, abstractmethod

class Extractor(ABC):
    @abstractmethod
    def rows(self): ...

class CsvExtractor(Extractor):
    def __init__(self, path): self.path = path
    def rows(self):
        with open(self.path, newline="") as f:
            yield from csv.DictReader(f)

class JsonlExtractor(Extractor):
    def __init__(self, path): self.path = path
    def rows(self):
        with open(self.path) as f:
            for line in f:
                yield json.loads(line)

sources = [CsvExtractor("orders.csv"),
           JsonlExtractor("events.jsonl")]
for source in sources:
    n = sum(1 for _ in source.rows())   # same call, any source
    print(type(source).__name__, n)

try:
    Extractor()
except TypeError as e:
    print(e)

CsvExtractor 15
JsonlExtractor 12
Can't instantiate abstract class Extractor without an implementation for abstract method 'rows'