Template Method Workflow Steps Base Class in Python

from abc import ABC, abstractmethod


class DataPipeline(ABC):
    """Template Method pattern: defines a workflow skeleton."""

    def run(self):
        """Template method - defines the algorithm's structure."""
        result = {"extracted": False, "transformed": False, "loaded": False}
        raw_data = self._extract()
        result["extracted"] = True
        transformed = self._transform(raw_data)
        result["transformed"] = True
        self._load(transformed)
        result["loaded"] = True
        return result

    @abstractmethod
    def _extract(self):
        pass

    @abstractmethod
    def _transform(self, data):
        pass

    @abstractmethod
    def _load(self, data):
        pass


class CSVToDatabasePipeline(DataPipeline):
    """Concrete implementation for CSV to DB workflow."""

    def _extract(self):
        print("Extracting data from CSV...")
        return ["alice@example.com", "bob@example.com"]

    def _transform(self, data):
        print("Transforming emails...")
        return [email.strip().lower() for email in data]

    def _load(self, data):
        print(f"Loading {len(data)} rows into database...")
        for email in data:
            print(f"  - INSERT {email}")


class APIValidationPipeline(DataPipeline):
    """Concrete implementation for API validation workflow."""

    def _extract(self):
        print("Fetching API responses...")
        return {"status": "ok", "code": 200}

    def _transform(self, data):
        print("Validating response...")
        return data["status"] == "ok"

    def _load(self, data):
        print(f"Saving validation result: {data}")


if __name__ == "__main__":
    print("=== CSV Pipeline ===")
    csv_result = CSVToDatabasePipeline().run()
    print(f"Result: {csv_result}")

    print("\n=== API Pipeline ===")
    api_result = APIValidationPipeline().run()
    print(f"Result: {api_result}")