Wzorce projektowe w Pythonie dla czystej architektury

Twórz utrwalane aplikacje Pythona z wykorzystaniem wzorców projektowych SOLID

Clean Architecture przekształciła sposób, w jaki programiści tworzą skalowalne, utrzymywalne aplikacje, podkreślając oddzielenie obowiązków i zarządzanie zależnościami.

W Pythonie te zasady łączą się z dynamiczną naturą języka, tworząc elastyczne, testowalne systemy, które ewoluują wraz z wymaganiami biznesowymi bez stania się długiem technicznym.

żywiołowy hal technologicznej konferencji

Zrozumienie Clean Architecture w Pythonie

Clean Architecture, wprowadzona przez Roberta C. Martina (Uncle Boba), organizuje oprogramowanie w koncentryczne warstwy, w których zależności wskazują wewnętrznie w stronę rdzennej logiki biznesowej. Ten wzorzec architektoniczny zapewnia, że kluczowe reguły biznesowe aplikacji pozostają niezależne od frameworków, baz danych i usług zewnętrznych.

Podstawowa filozofia

Podstawowy zasada jest prosta, ale potężna: logika biznesowa nie powinna zależeć od infrastruktury. Twoje obiekty domeny, przypadki użycia i reguły biznesowe powinny działać niezależnie od tego, czy korzystasz z PostgreSQL czy MongoDB, FastAPI czy Flask, AWS czy Azure.

W Pythonie ta filozofia idealnie pasuje do natury języka “duck typing” i programowania opartego na protokołach, umożliwiając czyste oddzielenie bez ceremonii wymaganej w językach statycznie typowanych.

Cztery warstwy Clean Architecture

Warstwa obiektów (Domena): Puri czyste obiekty biznesowe z regułami biznesowymi obowiązującymi w całym przedsiębiorstwie. Są to POJO (Plain Old Python Objects) bez żadnych zależności zewnętrznych.

Warstwa przypadków użycia (Aplikacja): Reguły biznesowe aplikacji, które koordynują przepływ danych między obiektami a usługami zewnętrznymi.

Warstwa adapterów interfejsów: Konwertuje dane w formacie najbardziej wygodnym dla przypadków użycia i obiektów, oraz w formacie wymaganym przez agencje zewnętrzne.

Warstwa frameworków i sterowników: Wszystkie szczegóły zewnętrzne, takie jak bazy danych, frameworki sieciowe i zewnętrzne API.

Zasady SOLID w Pythonie

Zasady SOLID stanowią fundament czystej architektury. Przeanalizujmy, jak każda z nich manifestuje się w Pythonie. Dla pełnego przeglądu wzorców projektowych w Pythonie zobacz Python Design Patterns Guide.

Zasada jednej odpowiedzialności (SRP)

Każda klasa powinna mieć jeden powód do zmiany:

# Zły: Wiele odpowiedzialności
class UserManager:
    def create_user(self, user_data):
        # Tworzenie użytkownika
        pass
    
    def send_welcome_email(self, user):
        # Wysyłanie e-maila
        pass
    
    def log_creation(self, user):
        # Logowanie do pliku
        pass

# Dobry: Oddzielone odpowiedzialności
class UserService:
    def __init__(self, repository, email_service, logger):
        self.repository = repository
        self.email_service = email_service
        self.logger = logger
    
    def create_user(self, user_data):
        user = User(**user_data)
        self.repository.save(user)
        self.email_service.send_welcome(user)
        self.logger.info(f"Użytkownik utworzony: {user.id}")
        return user

Zasada otwartej/ zamkniętej (OCP)

Jednostki oprogramowania powinny być otwarte na rozszerzenie, ale zamknięte na modyfikację:

from abc import ABC, abstractmethod
from typing import Protocol

# Używanie Protokołu (Python 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # Logika karty kredytowej
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # Logika PayPal
        return True

# Łatwe rozszerzanie bez modyfikacji istniejącego kodu
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Logika kryptowalut
        return True

Zasada podstawienia Liskova (LSP)

Obiekty powinny być zamieniane na ich podtypy bez naruszania działania programu:

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def save(self, key: str, value: str) -> None:
        pass
    
    @abstractmethod
    def get(self, key: str) -> str:
        pass

class PostgreSQLStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Implementacja PostgreSQL
        pass
    
    def get(self, key: str) -> str:
        # Implementacja PostgreSQL
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Implementacja Redis
        pass
    
    def get(self, key: str) -> str:
        # Implementacja Redis
        return ""

# Oba mogą być używane zamiennie
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

Zasada segregacji interfejsów (ISP)

Klienci nie powinni być zmuszeni do zależności od interfejsów, których nie używają:

# Zły: Gruby interfejs
class Worker(ABC):
    @abstractmethod
    def work(self): pass
    
    @abstractmethod
    def eat(self): pass
    
    @abstractmethod
    def sleep(self): pass

# Dobry: Oddzielone interfejsy
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("Pracuję")
    
    def eat(self) -> None:
        print("Jem")

class Robot:
    def work(self) -> None:
        print("Pracuję")
    # Nie potrzebna metoda eat

Zasada odwrócenia zależności (DIP)

Moduły wysokiego poziomu nie powinny zależeć od modułów niskiego poziomu. Oba powinny zależeć od abstrakcji:

from typing import Protocol

# Abstrakcja
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# Moduł niskiego poziomu
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # Implementacja SMTP
        pass

# Moduł wysokiego poziomu zależny od abstrakcji
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender
    
    def register(self, email: str, name: str):
        # Logika rejestracji
        self.email_sender.send(
            to=email,
            subject="Witaj!",
            body=f"Witaj {name}"
        )

Wzorzec Repository: Abstrahowanie dostępu do danych

Wzorzec Repository oferuje interfejs podobny do kolekcji do dostępu do obiektów domeny, ukrywając szczegóły przechowywania danych.

Podstawowa implementacja Repository

from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID
    email: str
    name: str
    is_active: bool = True

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass
    
    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]:
        pass
    
    @abstractmethod
    def list_all(self) -> List[User]:
        pass
    
    @abstractmethod
    def delete(self, user_id: UUID) -> bool:
        pass

Implementacja SQLAlchemy

from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'
    
    id = Column(PGUUID(as_uuid=True), primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session
    
    def save(self, user: User) -> User:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
        self.session.add(user_model)
        self.session.commit()
        return user
    
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).first()
        
        if not user_model:
            return None
        
        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )
    
    def get_by_email(self, email: str) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.email == email
        ).first()
        
        if not user_model:
            return None
        
        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )
    
    def list_all(self) -> List[User]:
        users = self.session.query(UserModel).all()
        return [
            User(
                id=u.id,
                email=u.email,
                name=u.name,
                is_active=u.is_active
            )
            for u in users
        ]
    
    def delete(self, user_id: UUID) -> bool:
        result = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).delete()
        self.session.commit()
        return result > 0

Repository w pamięci dla testów

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users: dict[UUID, User] = {}
    
    def save(self, user: User) -> User:
        self.users[user.id] = user
        return user
    
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        return self.users.get(user_id)
    
    def get_by_email(self, email: str) -> Optional[User]:
        for user in self.users.values():
            if user.email == email:
                return user
        return None
    
    def list_all(self) -> List[User]:
        return list(self.users.values())
    
    def delete(self, user_id: UUID) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

Warstwa usług: Koordynowanie logiki biznesowej

Warstwa usług implementuje przypadki użycia i koordynuje przepływ między repozytoriami, usługami zewnętrznymi i logiką domeny.

from typing import Optional
from uuid import uuid4

class UserAlreadyExistsError(Exception):
    pass

class UserNotFoundError(Exception):
    pass

class UserService:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailSender,
        event_publisher: 'EventPublisher'
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.event_publisher = event_publisher
    
    def register_user(self, email: str, name: str) -> User:
        # Sprawdzenie, czy użytkownik istnieje
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"Użytkownik z e-mailem {email} już istnieje")
        
        # Utworzenie nowego użytkownika
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )
        
        # Zapisanie w repozytorium
        user = self.user_repository.save(user)
        
        # Wysłanie powitalnego e-maila
        self.email_service.send(
            to=user.email,
            subject="Witaj!",
            body=f"Witaj {user.name}, witaj na naszej platformie!"
        )
        
        # Wyślij zdarzenie
        self.event_publisher.publish('user.registered', {
            'user_id': str(user.id),
            'email': user.email
        })
        
        return user
    
    def deactivate_user(self, user_id: UUID) -> User:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"Użytkownik {user_id} nie znaleziony")
        
        user.is_active = False
        user = self.user_repository.save(user)
        
        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })
        
        return user

Iniekcja zależności w Pythonie

Dynamiczna natura Pythona sprawia, że iniekcja zależności jest prosta bez konieczności korzystania z ciężkich frameworków.

Iniekcja przez konstruktor

class OrderService:
    def __init__(
        self,
        order_repository: 'OrderRepository',
        payment_processor: PaymentProcessor,
        notification_service: 'NotificationService'
    ):
        self.order_repository = order_repository
        self.payment_processor = payment_processor
        self.notification_service = notification_service
    
    def place_order(self, order_data: dict):
        # Użycie wstrzykiwanych zależności
        pass

Prosty kontener zależności

from typing import Dict, Type, Callable, Any

class Container:
    def __init__(self):
        self._services: Dict[Type, Callable] = {}
        self._singletons: Dict[Type, Any] = {}
    
    def register(self, interface: Type, factory: Callable):
        self._services[interface] = factory
    
    def register_singleton(self, interface: Type, instance: Any):
        self._singletons[interface] = instance
    
    def resolve(self, interface: Type):
        if interface in self._singletons:
            return self._singletons[interface]
        
        factory = self._services.get(interface)
        if factory:
            return factory(self)
        
        raise ValueError(f"Nie znaleziono rejestracji dla {interface}")

# Użycie
def create_container() -> Container:
    container = Container()
    
    # Rejestrowanie usług
    container.register_singleton(
        Session,
        sessionmaker(bind=create_engine('postgresql://...'))()
    )
    
    container.register(
        UserRepository,
        lambda c: SQLAlchemyUserRepository(c.resolve(Session))
    )
    
    container.register(
        EmailSender,
        lambda c: SMTPEmailSender()
    )
    
    container.register(
        UserService,
        lambda c: UserService(
            c.resolve(UserRepository),
            c.resolve(EmailSender),
            c.resolve(EventPublisher)
        )
    )
    
    return container

Architektura Sześciokątna (Porty i Adaptery)

Architektura Sześciokątna umieszcza logikę biznesową w centrum, a adaptery zajmują się komunikacją zewnętrzną.

Definiowanie portów (interfejsów)

# Wejściowy port (główny)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Wyjściowy port (pomocniczy)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...
    
    def find_by_email(self, email: str) -> Optional[User]:
        ...

Implementacja adapterów

from pydantic import BaseModel, EmailStr

# Wejściowy adapter (API REST)
from fastapi import FastAPI, Depends, HTTPException

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str

class CreateUserResponse(BaseModel):
    id: str
    email: str
    name: str

app = FastAPI()

@app.post("/users", response_model=CreateUserResponse)
def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(get_user_service)
):
    try:
        user = user_service.register_user(
            email=request.email,
            name=request.name
        )
        return CreateUserResponse(
            id=str(user.id),
            email=user.email,
            name=user.name
        )
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=400, detail=str(e))

# Wyjściowy adapter (baza danych)
# Już zaimplementowany jako SQLAlchemyUserRepository

Wzorce projektowe oparte na domenie

Obiekty wartości

Niezmienne obiekty zdefiniowane przez swoje atrybuty:

from dataclasses import dataclass
from typing import Pattern
import re

@dataclass(frozen=True)
class Email:
    value: str
    
    EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
    
    def __post_init__(self):
        if not self.EMAIL_PATTERN.match(self.value):
            raise ValueError(f"Nieprawidłowy e-mail: {self.value}")
    
    def __str__(self):
        return self.value

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str
    
    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Kwota nie może być ujemna")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Nieobsługiwana waluta: {self.currency}")
    
    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Nie można dodać różnych walut")
        return Money(self.amount + other.amount, self.currency)

Agregaty

Klaster obiektów domeny traktowanych jako jednostka:

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class OrderItem:
    product_id: UUID
    quantity: int
    price: Money
    
    def total(self) -> Money:
        return Money(
            self.price.amount * self.quantity,
            self.price.currency
        )

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    items: List[OrderItem] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)
    
    def add_item(self, product_id: UUID, quantity: int, price: Money):
        item = OrderItem(product_id, quantity, price)
        self.items.append(item)
    
    def remove_item(self, product_id: UUID):
        self.items = [
            item for item in self.items 
            if item.product_id != product_id
        ]
    
    def total(self) -> Money:
        if not self.items:
            return Money(0, "USD")
        
        return sum(
            (item.total() for item in self.items),
            Money(0, self.items[0].price.currency)
        )
    
    def confirm(self):
        if not self.items:
            raise ValueError("Nie można potwierdzić pustego zamówienia")
        if self.status != "pending":
            raise ValueError("Zamówienie już przetworzone")
        self.status = "confirmed"

Zdarzenia domeny

Zdarzenia domeny umożliwiają luźne sprzężenie między komponentami i wspierają architektury oparte na zdarzeniach. Dla systemów opartych na zdarzeniach w dużych skalach rozważ implementację strumieni zdarzeń z usługami takimi jak AWS Kinesis — zobacz Wbudowanie mikroserwisów opartych na zdarzeniach z użyciem AWS Kinesis dla szczegółowego przewodnika.

from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.now)

@dataclass
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total: Money

class EventPublisher:
    def __init__(self):
        self._handlers: Dict[Type, List[Callable]] = {}
    
    def subscribe(self, event_type: Type, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
    
    def publish(self, event: DomainEvent):
        event_type = type(event)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            handler(event)

Nowoczesne funkcje Pythona dla czystej architektury

Nowoczesne funkcje Pythona sprawiają, że implementowanie czystej architektury jest bardziej eleganckie i typowo bezpieczne. Jeśli potrzebujesz szybkiego odniesienia do składni i funkcji Pythona, sprawdź Python Cheatsheet.

Wskazówki typowe i protokoły

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict:
        ...
    
    @classmethod
    def from_dict(cls, data: dict) -> 'Serializable':
        ...

def serialize(obj: Serializable) -> dict:
    return obj.to_dict()

Pydantic do walidacji

from pydantic import BaseModel, Field, validator
from typing import Optional

class CreateUserDTO(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: Optional[int] = Field(None, ge=0, le=150)
    
    @validator('name')
    def name_must_not_contain_numbers(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Imię nie może zawierać cyfr')
        return v
    
    class Config:
        frozen = True  # Ustawienie jako niemutowalne

Async/Await dla operacji we/wy

Składnia async/await w Pythonie jest szczególnie potężna w przypadku operacji we/wy w czystej architekturze, umożliwiając niesynchronizowane interakcje z bazami danych i zewnętrznymi usługami. Podczas wdrażania aplikacji Pythona na platformach bezserwerowych, zrozumienie cech wydajności staje się kluczowe – zobacz AWS lambda performance: JavaScript vs Python vs Golang dla wskazówek dotyczących optymalizacji funkcji Pythona w architekturze bezserwerowej.

from typing import List
import asyncio

class AsyncUserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

class AsyncUserService:
    def __init__(self, repository: AsyncUserRepository):
        self.repository = repository
    
    async def register_user(self, email: str, name: str) -> User:
        user = User(id=uuid4(), email=email, name=name)
        return await self.repository.save(user)
    
    async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
        tasks = [self.repository.get_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

Najlepsze praktyki dotyczące struktury projektu

Poprawna organizacja projektu jest kluczowa dla utrzymania czystej architektury. Przed ustawieniem struktury projektu upewnij się, że korzystasz z wirtualnych środowisk Pythona do izolacji zależności. venv Cheatsheet zawiera wszystko, co musisz wiedzieć na temat zarządzania środowiskami wirtualnymi. Dla nowoczesnych projektów Pythona rozważ użycie uv - Nowy menedżer pakietów, projektów i środowisk Pythona, który oferuje szybsze zarządzanie pakietami i konfigurację projektu.

moja_aplikacja/
├── domena/                 # Reguły biznesowe firmy
│   ├── __init__.py
│   ├── encje/
│   │   ├── __init__.py
│   │   ├── użytkownik.py
│   │   └── zamówienie.py
│   ├── wartości/
│   │   ├── __init__.py
│   │   ├── e-mail.py
│   │   └── pieniądze.py
│   ├── zdarzenia/
│   │   ├── __init__.py
│   │   └── zdarzenia_użytkownika.py
│   └── wyjątki.py
│
├── aplikacja/            # Reguły biznesowe aplikacji
│   ├── __init__.py
│   ├── przypadki_użycia/
│   │   ├── __init__.py
│   │   ├── utwórz_użytkownika.py
│   │   └── złożenie_zamówienia.py
│   ├── usługi/
│   │   ├── __init__.py
│   │   └── usługa_użytkownika.py
│   └── porty/
│       ├── __init__.py
│       ├── repozytoria.py
│       └── usługi_zewnętrzne.py
│
├── infrastruktura/         # Interfejsy zewnętrzne
│   ├── __init__.py
│   ├── trwałość/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── modele.py
│   │   │   └── repozytoria.py
│   │   └── mongodb/
│   │       └── repozytoria.py
│   ├── komunikacja/
│   │   ├── __init__.py
│   │   └── publisher_rabbitmq.py
│   ├── usługi_zewnętrzne/
│   │   ├── __init__.py
│   │   └── usługa_e-mail.py
│   └── konfiguracja.py
│
├── prezentacja/           # Warstwa UI/API
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── zależności.py
│   │   ├── trasy/
│   │   │   ├── __init__.py
│   │   │   ├── użytkownicy.py
│   │   │   └── zamówienia.py
│   │   └── schematy/
│   │       ├── __init__.py
│   │       └── schematy_użytkownika.py
│   └── cli/
│       └── polecenia.py
│
├── testy/
│   ├── testy_jednostkowe/
│   ├── testy_integracyjne/
│   └── testy_e2e/
│
├── main.py                 # Punkt wejścia aplikacji
├── kontener.py            # Konfiguracja iniekcji zależności
├── pyproject.toml
└── README.md

Testowanie czystej architektury

Testowanie logiki domeny

import pytest
from uuid import uuid4

def test_utworzenie_użytkownika():
    użytkownik = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert użytkownik.email == "test@example.com"
    assert użytkownik.is_active is True

def test_obliczenie_sumy_zamówienia():
    zamówienie = Order(id=uuid4(), customer_id=uuid4())
    zamówienie.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    zamówienie.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert zamówienie.total().amount == 25.0

Testowanie integracji z repozytorium

@pytest.fixture
def repozytorium_w_pamieci():
    return InMemoryUserRepository()

def test_zapisanie_i_pobranie_użytkownika(repozytorium_w_pamieci):
    użytkownik = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    
    zapisany_użytkownik = repozytorium_w_pamieci.save(użytkownik)
    pobrany_użytkownik = repozytorium_w_pamieci.get_by_id(użytkownik.id)
    
    assert pobrany_użytkownik is not None
    assert pobrany_użytkownik.email == użytkownik.email

Testowanie warstwy usług

from unittest.mock import Mock

def test_rejestracja_użytkownika():
    # Przygotowanie
    mock_repozytorium = Mock(spec=UserRepository)
    mock_repozytorium.get_by_email.return_value = None
    mock_repozytorium.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )
    
    mock_email = Mock(spec=EmailSender)
    mock_zdarzenia = Mock(spec=EventPublisher)
    
    usługa = UserService(mock_repozytorium, mock_email, mock_zdarzenia)
    
    # Wykonanie
    użytkownik = usługa.register_user("test@example.com", "Test")
    
    # Sprawdzenie
    assert użytkownik.email == "test@example.com"
    mock_repozytorium.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_zdarzenia.publish.assert_called_once()

Typowe pułapki i sposób ich unikania

Nadmierny projekt

Nie implementuj czystej architektury dla prostych aplikacji CRUD. Zacznij od prostego rozwiązania i refaktoryzuj, gdy złożoność rośnie.

Wyciekające abstrakcje

Upewnij się, że obiekty domeny nie zawierają adnotacji do bazy danych ani kodu specyficznych dla frameworków:

# Zły
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Framework wyciekający do domeny

# Dobry
@dataclass
class User:
    id: UUID  # Puro obiekt domeny

Zależności cykliczne

Używaj iniekcji zależności i interfejsów, aby przerwać zależności cykliczne między warstwami.

Ignorowanie kontekstu

Czysta architektura nie jest jednoznaczna dla wszystkich. Dostosuj ściśleść warstw w zależności od rozmiaru projektu i doświadczenia zespołu.

Przydatne linki