Przegląd
Ta implementacja opiera się na klasach abstrakcyjnych i dziedziczeniu. Moduł abc daje wbudowaną obsługę metod abstrakcyjnych, więc tradycyjne podejście jest proste i zgodne z duchem Pythona.
Zalety tego podejścia
- Jawna struktura algorytmu: metody abstrakcyjne mówią dokładnie, co musi zaimplementować podklasa.
- Jasne umowy: klasa bazowa od razu pokazuje szkielet algorytmu.
- Narzędzia do sprawdzania typów: IDE i analiza typów potrafią zweryfikować implementacje.
- Znany wzorzec OOP: działa jak klasyczna metoda szablonowa.
- Pythonowy styl: naturalnie korzysta z wbudowanego modułu
abc.
Tradycyjne podejście: dziedziczenie z metodami abstrakcyjnymi
Klasa bazowa z metodami abstrakcyjnymi
from abc import ABC, abstractmethod
class DataProcessorBase(ABC):
"""
Abstrakcyjna klasa bazowa definiująca wzorzec Metody Szablonowej.
Metoda execute() jest metodą szablonową, która definiuje szkielet algorytmu.
Podklasy muszą implementować dostosowywalne kroki.
"""
def execute(self, data: str) -> None:
"""
Metoda szablonowa: definiuje szkielet algorytmu.
Kroki są zdefiniowane, ale podklasy je dostosowują:
1. Waliduj wejście
2. Przetwórz dane
3. Zapisz wyniki
"""
print("Rozpoczynanie potoku przetwarzania danych...")
# Krok 1: Walidacja
validated = self.validate(data)
if not validated:
print("Walidacja nie powiodła się. Przerywanie.")
return
# Krok 2: Przetwarzanie
processed = self.process(validated)
print("Przetwarzanie ukończone.")
# Krok 3: Zapis
self.save(processed)
print("Potok zakończony pomyślnie.\n")
# Metody abstrakcyjne: podklasy muszą je implementować
@abstractmethod
def validate(self, data: str) -> str:
"""Waliduj wejście. Zwróć pusty string jeśli nieprawidłowy."""
pass
@abstractmethod
def process(self, data: str) -> str:
"""Przetwórz walidowane dane."""
pass
@abstractmethod
def save(self, data: str) -> None:
"""Zapisz przetworzone dane."""
pass
Konkretne implementacje
class CSVProcessor(DataProcessorBase):
"""Implementacja do przetwarzania danych CSV."""
def validate(self, data: str) -> str:
"""Waliduj format CSV."""
print(" [CSV Walidacja] Sprawdzanie formatu CSV...")
return data if data else ""
def process(self, data: str) -> str:
"""Parsuj i wyczyść CSV."""
print(" [CSV Przetwarzanie] Parsowanie i czyszczenie CSV...")
return f"{data} (wyczyszczone)"
def save(self, data: str) -> None:
"""Napisz do pliku CSV."""
print(f" [CSV Zapis] Pisanie do pliku CSV: {data}")
class JSONProcessor(DataProcessorBase):
"""Implementacja do przetwarzania danych JSON."""
def validate(self, data: str) -> str:
"""Waliduj składnię JSON."""
print(" [JSON Walidacja] Walidowanie składni JSON...")
if not data or data[0] != '{':
return ""
return data
def process(self, data: str) -> str:
"""Parsuj i formatuj JSON."""
print(" [JSON Przetwarzanie] Parsowanie i formatowanie JSON...")
return f"{data} (sformatowane)"
def save(self, data: str) -> None:
"""Napisz do pliku JSON."""
print(f" [JSON Zapis] Pisanie do pliku JSON: {data}")
Przykład tradycyjnego użycia
def main():
print("=== Wzorzec Metoda Szablonowa (Dziedziczenie) ===\n")
# Potok CSV
print("1. Potok CSV:")
csv_processor = CSVProcessor()
csv_processor.execute("imię,wiek,email")
# Potok JSON
print("2. Potok JSON:")
json_processor = JSONProcessor()
json_processor.execute('{"klucz": "wartość"}')
if __name__ == "__main__":
main()
Wyjście
=== Wzorzec Metoda Szablonowa (Dziedziczenie) ===
1. Potok CSV:
Rozpoczynanie potoku przetwarzania danych...
[CSV Walidacja] Sprawdzanie formatu CSV...
[CSV Przetwarzanie] Parsowanie i czyszczenie CSV...
Przetwarzanie ukończone.
[CSV Zapis] Pisanie do pliku CSV: imię,wiek,email (wyczyszczone)
Potok zakończony pomyślnie.
2. Potok JSON:
Rozpoczynanie potoku przetwarzania danych...
[JSON Walidacja] Walidowanie składni JSON...
[JSON Przetwarzanie] Parsowanie i formatowanie JSON...
Przetwarzanie ukończone.
[JSON Zapis] Pisanie do pliku JSON: {"klucz": "wartość"} (sformatowane)
Potok zakończony pomyślnie.
Kluczowe punkty: dziedziczenie
Abstrakcyjna klasa bazowa
Klasa DataProcessorBase używa ABC i @abstractmethod do wymuszenia implementacji wymaganych metod. Python uniemożliwia instancjonowanie klas abstrakcyjnych.
Metoda szablonowa
execute() jest metodą szablonową i organizuje wszystkie kroki. Podklasy nie mogą jej zastąpić, więc struktura algorytmu pozostaje spójna.
Metody abstrakcyjne
Metody validate(), process() i save() są abstrakcyjne. Podklasa musi zaimplementować wszystkie trzy, inaczej sama pozostaje abstrakcyjna.
Jasne umowy
Klasa bazowa jasno mówi, co mają robić podklasy. Narzędzia takie jak mypy potrafią zweryfikować implementacje.
Nowoczesne podejście: callables (kompozycja)
Python traktuje funkcje jak zwykłe wartości, więc możesz przekazać callables przez konstruktor. Unikasz dziedziczenia, a struktura algorytmu zostaje ta sama.
Klasa algorytmu z callable
from typing import Callable
class DataProcessor:
"""
Wzorzec metody szablonowej używający funkcji callable.
Dostosowywalne kroki są przekazywane w konstruktorze,
umożliwiając elastyczne implementacje algorytmu bez dziedziczenia.
"""
def __init__(
self,
validate: Callable[[str], str],
process: Callable[[str], str],
save: Callable[[str], None]
):
"""
Inicjalizuj z niestandardowymi krokami algorytmu.
Args:
validate: Funkcja, która waliduje wejście, zwraca pusty string jeśli nieprawidłowy
process: Funkcja, która przetwarza wejście
save: Funkcja, która zapisuje przetworzone dane
"""
self._validate = validate
self._process = process
self._save = save
def execute(self, data: str) -> None:
"""
Metoda szablonowa: definiuje szkielet algorytmu.
Kroki są dostosowywalne, ale ogólny przepływ jest ustalony:
1. Waliduj wejście
2. Przetwórz dane
3. Zapisz wyniki
"""
print("Rozpoczynanie potoku przetwarzania danych...")
# Krok 1: Walidacja
validated = self._validate(data)
if not validated:
print("Walidacja nie powiodła się. Przerywanie.")
return
# Krok 2: Przetwarzanie
processed = self._process(validated)
print("Przetwarzanie ukończone.")
# Krok 3: Zapis
self._save(processed)
print("Potok zakończony pomyślnie.\n")
Implementacje oparte na funkcjach
class CSVProcessorFn:
"""Implementacja do przetwarzania danych CSV (oparta na funkcjach)."""
@staticmethod
def validate(data: str) -> str:
"""Waliduj format CSV."""
print(" [CSV Walidacja] Sprawdzanie formatu CSV...")
return data if data else ""
@staticmethod
def process(data: str) -> str:
"""Parsuj i wyczyść CSV."""
print(" [CSV Przetwarzanie] Parsowanie i czyszczenie CSV...")
return f"{data} (wyczyszczone)"
@staticmethod
def save(data: str) -> None:
"""Napisz do pliku CSV."""
print(f" [CSV Zapis] Pisanie do pliku CSV: {data}")
class JSONProcessorFn:
"""Implementacja do przetwarzania danych JSON (oparta na funkcjach)."""
@staticmethod
def validate(data: str) -> str:
"""Waliduj składnię JSON."""
print(" [JSON Walidacja] Walidowanie składni JSON...")
if not data or data[0] != '{':
return ""
return data
@staticmethod
def process(data: str) -> str:
"""Parsuj i formatuj JSON."""
print(" [JSON Przetwarzanie] Parsowanie i formatowanie JSON...")
return f"{data} (sformatowane)"
@staticmethod
def save(data: str) -> None:
"""Napisz do pliku JSON."""
print(f" [JSON Zapis] Pisanie do pliku JSON: {data}")
Użycie funkcji lambda
class CustomProcessor:
"""Utwórz procesory z funkcjami lambda."""
@staticmethod
def create_with_lambdas() -> DataProcessor:
"""Zbuduj procesor używając wbudowanych wyrażeń lambda."""
return DataProcessor(
# Walidacja: odrzuć jeśli zawiera znaki specjalne
validate=lambda data: (
print(" [Lambda Walidacja] Sprawdzanie znaków specjalnych...") or
("" if "!" in data else data)
),
# Przetwarzanie: konwertuj na wielkie litery
process=lambda data: (
print(" [Lambda Przetwarzanie] Konwersja na wielkie litery...") or
data.upper()
),
# Zapis: wydruk na konsoli
save=lambda data: print(f" [Lambda Zapis] Wynik: {data}")
)
Metody instancji jako callbacki
class EmailProcessor:
"""Przetwarzaj i waliduj dane email."""
def __init__(self, domain_whitelist: list):
self.domain_whitelist = domain_whitelist
def validate(self, data: str) -> str:
"""Waliduj format email i domenę."""
print(" [Email Walidacja] Sprawdzanie formatu email...")
if "@" not in data:
return ""
domain = data.split("@")[1]
return data if domain in self.domain_whitelist else ""
def process(self, data: str) -> str:
"""Wyczyść i znormalizuj email."""
print(" [Email Przetwarzanie] Normalizowanie email...")
return data.lower().strip()
def save(self, data: str) -> None:
"""Napisz do bazy danych email."""
print(f" [Email Zapis] Przechowywanie w bazie danych: {data}")
Przykład użycia kompozycji
def main():
print("=== Wzorzec Metoda Szablonowa (Callable) ===\n")
# Potok CSV
print("1. Potok CSV:")
csv_pipeline = DataProcessor(
CSVProcessorFn.validate,
CSVProcessorFn.process,
CSVProcessorFn.save
)
csv_pipeline.execute("imię,wiek,email")
# Potok JSON
print("2. Potok JSON:")
json_pipeline = DataProcessor(
JSONProcessorFn.validate,
JSONProcessorFn.process,
JSONProcessorFn.save
)
json_pipeline.execute('{"klucz": "wartość"}')
# Potok Oparty na Lambda
print("3. Potok Oparty na Lambda:")
custom_pipeline = CustomProcessor.create_with_lambdas()
custom_pipeline.execute("cześć")
# Przykład Nieudanej Walidacji
print("4. Przykład Nieudanej Walidacji:")
custom_pipeline.execute("niepoprawne!")
# Procesor Email z Metodami Instancji
print("5. Procesor Email:")
email_processor = EmailProcessor(
domain_whitelist=["example.com", "company.com"]
)
email_pipeline = DataProcessor(
email_processor.validate,
email_processor.process,
email_processor.save
)
email_pipeline.execute("User@Example.com")
email_pipeline.execute("user@invalid.com")
if __name__ == "__main__":
main()
Wyjście (callable)
=== Wzorzec Metoda Szablonowa (Callable) ===
1. Potok CSV:
Rozpoczynanie potoku przetwarzania danych...
[CSV Walidacja] Sprawdzanie formatu CSV...
[CSV Przetwarzanie] Parsowanie i czyszczenie CSV...
Przetwarzanie ukończone.
[CSV Zapis] Pisanie do pliku CSV: imię,wiek,email (wyczyszczone)
Potok zakończony pomyślnie.
2. Potok JSON:
Rozpoczynanie potoku przetwarzania danych...
[JSON Walidacja] Walidowanie składni JSON...
[JSON Przetwarzanie] Parsowanie i formatowanie JSON...
Przetwarzanie ukończone.
[JSON Zapis] Pisanie do pliku JSON: {"klucz": "wartość"} (sformatowane)
Potok zakończony pomyślnie.
3. Potok Oparty na Lambda:
Rozpoczynanie potoku przetwarzania danych...
[Lambda Walidacja] Sprawdzanie znaków specjalnych...
[Lambda Przetwarzanie] Konwersja na wielkie litery...
Przetwarzanie ukończone.
[Lambda Zapis] Wynik: CZEŚĆ
Potok zakończony pomyślnie.
4. Przykład Nieudanej Walidacji:
Rozpoczynanie potoku przetwarzania danych...
[Lambda Walidacja] Sprawdzanie znaków specjalnych...
Walidacja nie powiodła się. Przerywanie.
5. Procesor Email:
Rozpoczynanie potoku przetwarzania danych...
[Email Walidacja] Sprawdzanie formatu email...
[Email Przetwarzanie] Normalizowanie email...
Przetwarzanie ukończone.
[Email Zapis] Przechowywanie w bazie danych: user@example.com
Potok zakończony pomyślnie.
Rozpoczynanie potoku przetwarzania danych...
[Email Walidacja] Sprawdzanie formatu email...
Walidacja nie powiodła się. Przerywanie.
Kluczowe punkty: callable
Wstrzykiwanie zależności przez konstruktor
Konstruktor DataProcessor przyjmuje wszystkie dostosowywalne kroki od razu:
DataProcessor(validate, process, save)
Dzięki temu zależności są jawne, a testowanie z funkcjami mock prostsze.
Typy callable w Pythonie
Jako callable działa:
- Funkcja:
def my_func(data): ... - Lambda:
lambda data: ... - Metoda:
obj.method - Metoda statyczna:
ClassName.method - Obiekt z
__call__: dowolny obiekt z metodą__call__
Duck typing
Python nie wymaga jawnej implementacji interfejsu. Wystarczy, że obiekt da się wywołać z pasującą sygnaturą:
# Nawet niestandardowa klasa callable działa
class MyValidator:
def __call__(self, data: str) -> str:
return data if data else ""
processor = DataProcessor(
MyValidator(),
process_func,
save_func
)
Elastyczność w czasie wykonania
Zamiast dziedziczenia możesz zmienić zachowanie w czasie wykonania: przekazujesz inne funkcje albo modyfikujesz domknięcia.
Zaawansowany przykład: dekorowanie zachowania
def with_logging(name: str, fn: Callable) -> Callable:
"""Dekorator, aby dodać logowanie do dowolnego kroku algorytmu."""
def wrapper(data: str):
print(f"[{name}] Rozpoczynanie...")
result = fn(data)
print(f"[{name}] Ukończone.")
return result
return wrapper
# Użyj dekoratora, aby zawinąć kroki
logged_pipeline = DataProcessor(
with_logging("Walidacja", CSVProcessorFn.validate),
with_logging("Przetwarzanie", CSVProcessorFn.process),
with_logging("Zapis", CSVProcessorFn.save)
)
Porównanie: dziedziczenie i callable
| Aspekt | Dziedziczenie | Callable (kompozycja) |
|---|---|---|
| Struktura | Hierarchia klas | Prosta kompozycja |
| Powiązanie | Ścisłe, hierarchiczne | Luźne, elastyczne |
| Dziedziczenie | Wymagane | Nie wymagane |
| Testowanie | Wymaga podklas mock | Łatwe z funkcjami mock |
| Zmiany w czasie wykonania | Nie | Tak, trywialne |
| Złożoność | Hierarchie klas | Prosta kompozycja funkcji |
| Pythonowość | Wysoka (moduł abc) | Wysoka (funkcje pierwszej klasy) |
| Sprawdzanie typów | Jawna hierarchia typów | Wskazówki typów z Callable |
| Kod bazowy | Mniejszy (używa abc) | Minimalny |
| Rozszerzalność | Poprzez podklasy | Poprzez kompozycję funkcji |
Kiedy używać
Podejście dziedziczenia:
- Preferujesz tradycyjny projekt OOP.
- Chcesz jawnych umów wymuszanych przez moduł abc.
- Ścisłe sprzężenie jest do przyjęcia dla modelu domeny.
- Chcesz stan i zachowanie specyficzne dla podklasy.
- Zespół zna wzorce oparte na dziedziczeniu.
Podejście callable:
- Potrzebujesz elastyczności zachowania w czasie wykonania.
- Architektura opiera się na wtyczkach lub przypomina Strategy.
- Chcesz uniknąć hierarchii dziedziczenia.
- Testowanie z funkcjami mock jest priorytetem.
- Preferujesz styl programowania funkcyjnego.
Podsumowanie
Dziedziczenie czyni kontrakt jawnym i pozwala klasie bazowej pilnować przepływu. Obiekty wywoływalne zachowują ten sam przepływ przy mniejszej strukturze i większej swobodzie. Użyj dziedziczenia, gdy implementacje kroków należą do typu. Użyj obiektów wywoływalnych, gdy wywołujący składa algorytm z funkcji, które już ma.