BehavioralPythonverifiedVerified

Template Method Pattern in Python

Defines the skeleton of an algorithm in a base class, deferring certain steps to subclasses without changing the algorithm's overall structure.

How to Implement the Template Method Pattern in Python

1Step 1: Define the abstract template with hook methods

from abc import ABC, abstractmethod
import json


class DataMiner(ABC):
    def mine(self, source: str) -> list[str]:
        """Template method -- defines the algorithm skeleton."""
        raw = self.extract_data(source)
        parsed = self.parse_data(raw)
        filtered = self.filter_data(parsed)
        self.report_results(filtered)
        return filtered

    @abstractmethod
    def extract_data(self, source: str) -> str: ...

    @abstractmethod
    def parse_data(self, raw: str) -> list[str]: ...

    # Hook -- subclasses may override
    def filter_data(self, data: list[str]) -> list[str]:
        return data

    def report_results(self, data: list[str]) -> None:
        print(f"Mined {len(data)} records.")

2Step 2: Implement concrete miners that override steps

class CsvMiner(DataMiner):
    def extract_data(self, source: str) -> str:
        return f"CSV content of {source}"

    def parse_data(self, raw: str) -> list[str]:
        return [line for line in raw.split("\n") if line]


class JsonMiner(DataMiner):
    def extract_data(self, source: str) -> str:
        return '{"data": ["a", "b", "c"]}'

    def parse_data(self, raw: str) -> list[str]:
        return json.loads(raw)["data"]

    def filter_data(self, data: list[str]) -> list[str]:
        return [item for item in data if item != "b"]

3Step 3: Run the algorithm with different data sources

CsvMiner().mine("report.csv")
JsonMiner().mine("api/v1/data")

"""ETL Pipeline using Template Method with metrics and error handling."""

import logging
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Generic, TypeVar

logger = logging.getLogger(__name__)

TRaw = TypeVar("TRaw")
TTransformed = TypeVar("TTransformed")


# [step] Define pipeline metrics and record types
@dataclass
class PipelineMetrics:
    source: str = ""
    extracted_count: int = 0
    transformed_count: int = 0
    loaded_count: int = 0
    error_count: int = 0
    duration_ms: float = 0.0


# [step] Implement the abstract ETL pipeline template
class ETLPipeline(ABC, Generic[TRaw, TTransformed]):
    def __init__(self) -> None:
        self._metrics = PipelineMetrics()

    async def run(self, source: str) -> PipelineMetrics:
        """Template method -- orchestrates the full ETL flow."""
        start = time.monotonic()
        self._metrics = PipelineMetrics(source=source)

        await self.before_extract(source)
        raw_records = await self.extract(source)
        self._metrics.extracted_count = len(raw_records)

        transformed: list[TTransformed] = []
        for record in raw_records:
            try:
                result = await self.transform(record)
                if result is not None:
                    transformed.append(result)
            except Exception as exc:
                self._metrics.error_count += 1
                self.on_transform_error(record, exc)

        self._metrics.transformed_count = len(transformed)
        loaded = await self.load(transformed)
        self._metrics.loaded_count = loaded

        await self.after_load(transformed)
        self._metrics.duration_ms = (time.monotonic() - start) * 1000
        return self._metrics

    @abstractmethod
    async def extract(self, source: str) -> list[TRaw]: ...

    @abstractmethod
    async def transform(self, record: TRaw) -> TTransformed | None: ...

    @abstractmethod
    async def load(self, records: list[TTransformed]) -> int: ...

    # Hooks
    async def before_extract(self, source: str) -> None:
        pass

    async def after_load(self, records: list[TTransformed]) -> None:
        pass

    def on_transform_error(self, record: TRaw, error: Exception) -> None:
        logger.error("Transform error: %s (record=%s)", error, record)


# [step] Implement a concrete user import pipeline
@dataclass
class RawUser:
    name: str
    email: str
    created_at: str


@dataclass
class User:
    id: str
    name: str
    email: str
    created_at: str


class UserImportPipeline(ETLPipeline[RawUser, User]):
    def __init__(self) -> None:
        super().__init__()
        self.db: list[User] = []

    async def extract(self, source: str) -> list[RawUser]:
        return [
            RawUser("Alice", "[email protected]", "2024-01-15"),
            RawUser("Bob", "[email protected]", "2024-02-20"),
            RawUser("", "invalid", "bad-date"),
        ]

    async def transform(self, record: RawUser) -> User | None:
        if not record.name or "@" not in record.email:
            return None
        return User(
            id=str(uuid.uuid4()),
            name=record.name,
            email=record.email,
            created_at=record.created_at,
        )

    async def load(self, records: list[User]) -> int:
        self.db.extend(records)
        return len(records)

    async def after_load(self, records: list[User]) -> None:
        logger.info("Loaded %d users. Total in DB: %d", len(records), len(self.db))

Template Method Pattern Architecture

hourglass_empty

Rendering diagram...

lightbulb

Template Method Pattern in the Real World

“Consider a recipe for baking bread. The overall process—mix, knead, let rise, bake, cool—is fixed. But the specific flour blend, kneading technique, and baking temperature are decisions left to the baker. The cookbook provides the invariant sequence; individual bakers customize the steps that can vary without disrupting the overall process.”