Python-Entwurfsmuster für saubere Architektur

Erstellen Sie wartbare Python-Anwendungen mit SOLID-Entwurfsmustern

Clean Architecture hat die Art und Weise, wie Entwickler skalierbare, wartbare Anwendungen erstellen, revolutioniert, indem sie die Trennung von Verantwortlichkeiten und das Abhängigkeitsmanagement betonen.

In Python kombinieren sich diese Prinzipien mit der dynamischen Natur der Sprache, um flexible, testbare Systeme zu schaffen, die sich mit den Geschäftsanforderungen weiterentwickeln, ohne technische Schulden anzuhäufen.

vibrant tech conference hall

Verständnis von Clean Architecture in Python

Clean Architecture, eingeführt von Robert C. Martin (Uncle Bob), organisiert Software in konzentrischen Schichten, bei denen Abhängigkeiten nach innen zu den Kern-Geschäftslogiken zeigen. Dieses Architekturmodell stellt sicher, dass die kritischen Geschäftsregeln Ihrer Anwendung unabhängig von Frameworks, Datenbanken und externen Diensten bleiben.

Die Kernphilosophie

Das grundlegende Prinzip ist einfach, aber mächtig: Geschäftslogik sollte nicht von der Infrastruktur abhängen. Ihre Domänen-Entitäten, Use Cases und Geschäftsregeln sollten unabhängig davon funktionieren, ob Sie PostgreSQL oder MongoDB, FastAPI oder Flask, AWS oder Azure verwenden.

In Python passt sich diese Philosophie perfekt an die “Duck-Typing”- und protokollorientierte Programmierung der Sprache an, was eine saubere Trennung ohne den Aufwand ermöglicht, der in statisch typisierten Sprachen erforderlich ist.

Die vier Schichten der Clean Architecture

Entities Layer (Domain): Reine Geschäftsobjekte mit unternehmensweiten Geschäftsregeln. Dies sind POJOs (Plain Old Python Objects) ohne externe Abhängigkeiten.

Use Cases Layer (Application): Anwendungspezifische Geschäftsregeln, die den Datenfluss zwischen Entitäten und externen Diensten orchestrieren.

Interface Adapters Layer: Wandelt Daten zwischen dem für Use Cases und Entitäten am besten geeigneten Format und dem von externen Agenturen erforderlichen Format um.

Frameworks & Drivers Layer: Alle externen Details wie Datenbanken, Web-Frameworks und externe APIs.

SOLID-Prinzipien in Python

Die SOLID-Prinzipien bilden die Grundlage der Clean Architecture. Lassen Sie uns untersuchen, wie sich jedes Prinzip in Python manifestiert. Für einen umfassenden Überblick über Design-Patterns in Python siehe den Python Design Patterns Guide.

Single Responsibility Principle (SRP)

Jede Klasse sollte einen Grund zur Änderung haben:

# Schlecht: Mehrere Verantwortlichkeiten
class UserManager:
    def create_user(self, user_data):
        # Benutzer erstellen
        pass

    def send_welcome_email(self, user):
        # E-Mail senden
        pass

    def log_creation(self, user):
        # In Datei protokollieren
        pass

# Gut: Getrennte Verantwortlichkeiten
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"Benutzer erstellt: {user.id}")
        return user

Open/Closed Principle (OCP)

Software-Entitäten sollten für Erweiterungen offen, aber für Modifikationen geschlossen sein:

from abc import ABC, abstractmethod
from typing import Protocol

# Verwendung von Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # Kreditkarten-Logik
        return True

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

# Leicht erweiterbar ohne Modifikation des bestehenden Codes
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Kryptowährungs-Logik
        return True

Liskov Substitution Principle (LSP)

Objekte sollten durch ihre Subtypen ersetzbar sein, ohne das Programm zu beschädigen:

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:
        # PostgreSQL-Implementierung
        pass

    def get(self, key: str) -> str:
        # PostgreSQL-Implementierung
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Redis-Implementierung
        pass

    def get(self, key: str) -> str:
        # Redis-Implementierung
        return ""

# Beide können austauschbar verwendet werden
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

Interface Segregation Principle (ISP)

Clients sollten nicht gezwungen sein, auf Schnittstellen zu verzichten, die sie nicht verwenden:

# Schlecht: Fette Schnittstelle
class Worker(ABC):
    @abstractmethod
    def work(self): pass

    @abstractmethod
    def eat(self): pass

    @abstractmethod
    def sleep(self): pass

# Gut: Getrennte Schnittstellen
class Workable(Protocol):
    def work(self) -> None: ...

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

class Human:
    def work(self) -> None:
        print("Arbeiten")

    def eat(self) -> None:
        print("Essen")

class Robot:
    def work(self) -> None:
        print("Arbeiten")
    # Keine eat-Methode erforderlich

Dependency Inversion Principle (DIP)

Hochrangige Module sollten nicht von niederrangigen Modulen abhängen. Beide sollten von Abstraktionen abhängen:

from typing import Protocol

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

# Niederrangiges Modul
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # SMTP-Implementierung
        pass

# Hochrangiges Modul hängt von der Abstraktion ab
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender

    def register(self, email: str, name: str):
        # Registrierungslogik
        self.email_sender.send(
            to=email,
            subject="Willkommen!",
            body=f"Hallo {name}"
        )

Repository Pattern: Abstrahierung des Datenzugriffs

Das Repository Pattern bietet eine sammlungsähnliche Schnittstelle zum Zugriff auf Domänenobjekte und verbirgt die Details der Datenspeicherung.

Grundlegende Repository-Implementierung

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

SQLAlchemy-Implementierung

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

In-Memory-Repository für Tests

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

Service Layer: Orchestrierung der Geschäftslogik

Die Service Layer implementiert Use Cases und orchestriert den Datenfluss zwischen Repositories, externen Diensten und der Domänenlogik.

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:
        # Prüfen, ob Benutzer existiert
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"Benutzer mit der E-Mail {email} existiert bereits")

        # Neuen Benutzer erstellen
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )

        # In Repository speichern
        user = self.user_repository.save(user)

        # Willkommens-E-Mail senden
        self.email_service.send(
            to=user.email,
            subject="Willkommen!",
            body=f"Hallo {user.name}, willkommen auf unserer Plattform!"
        )

        # Ereignis veröffentlichen
        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"Benutzer {user_id} nicht gefunden")

        user.is_active = False
        user = self.user_repository.save(user)

        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })

        return user

Dependency Injection in Python

Die dynamische Natur von Python macht Dependency Injection einfach, ohne dass schwere Frameworks erforderlich sind.

Constructor Injection

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):
        # Injektierte Abhängigkeiten verwenden
        pass

Einfacher Dependency Container

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"Keine Registrierung gefunden für {interface}")

# Verwendung
def create_container() -> Container:
    container = Container()

    # Dienste registrieren
    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

Hexagonale Architektur (Ports und Adapters)

Die hexagonale Architektur platziert die Geschäftslogik im Zentrum, während Adapter die externe Kommunikation handhaben.

Definieren von Ports (Schnittstellen)

# Input Port (Primär)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Output Port (Sekundär)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...

    def find_by_email(self, email: str) -> Optional[User]:
        ...

Implementieren von Adaptern

from pydantic import BaseModel, EmailStr

# Input Adapter (REST API)
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))

# Output Adapter (Datenbank)
# Bereits implementiert als SQLAlchemyUserRepository

Domain-Driven Design Patterns

Value Objects

Unveränderliche Objekte, die durch ihre Attribute definiert sind:

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"Ungültige 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("Betrag kann nicht negativ sein")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Nicht unterstützte Währung: {self.currency}")

    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Kann verschiedene Währungen nicht addieren")
        return Money(self.amount + other.amount, self.currency)

Aggregates

Cluster von Domänenobjekten, die als einzelne Einheit behandelt werden:

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("Kann leere Bestellung nicht bestätigen")
        if self.status != "pending":
            raise ValueError("Bestellung bereits verarbeitet")
        self.status = "confirmed"

Domain Events

Domänenereignisse ermöglichen lockere Kopplung zwischen Komponenten und unterstützen ereignisgesteuerte Architekturen. Für produktionsreife ereignisgesteuerte Systeme sollten Sie die Implementierung von Event Streaming mit Diensten wie AWS Kinesis in Betracht ziehen - siehe Building Event-Driven Microservices with AWS Kinesis für eine detaillierte Anleitung.

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)

Moderne Python-Features für Clean Architecture

Python’s moderne Features machen die Implementierung von Clean Architecture eleganter und typsicher. Wenn Sie eine schnelle Referenz für Python-Syntax und Features benötigen, werfen Sie einen Blick auf den Python Cheatsheet.

Typ-Hinweise und Protokolle

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 für Validierung

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('Name cannot contain numbers')
        return v

    class Config:
        frozen = True  # Unveränderlich machen

Async/Await für I/O-Operationen

Python’s async/await-Syntax ist besonders leistungsfähig für I/O-gebundene Operationen in Clean Architecture, die nicht blockierende Interaktionen mit Datenbanken und externen Diensten ermöglichen. Beim Bereitstellen von Python-Anwendungen auf serverlosen Plattformen wird das Verständnis der Leistungsmerkmale entscheidend - siehe AWS Lambda-Leistung: JavaScript vs Python vs Golang für Einblicke in die Optimierung von Python-serverlosen Funktionen.

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]

Projektstruktur-Best Practices

Eine ordnungsgemäße Projektorganisation ist entscheidend für die Aufrechterhaltung einer sauberen Architektur. Bevor Sie Ihre Projektstruktur einrichten, stellen Sie sicher, dass Sie Python-Virtual-Umgebungen für die Isolierung von Abhängigkeiten verwenden. Der venv Cheatsheet behandelt alles, was Sie über das Verwalten von virtuellen Umgebungen wissen müssen. Für moderne Python-Projekte sollten Sie die Verwendung von uv - Neuer Python-Paket-, Projekt- und Umgebungsmanager in Betracht ziehen, der schnellere Paketverwaltung und Projektaufbau bietet.

my_application/
├── domain/                 # Unternehmensgeschäftsregeln
│   ├── __init__.py
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   └── money.py
│   ├── events/
│   │   ├── __init__.py
│   │   └── user_events.py
│   └── exceptions.py
│
├── application/            # Anwendungsgeschäftsregeln
│   ├── __init__.py
│   ├── use_cases/
│   │   ├── __init__.py
│   │   ├── create_user.py
│   │   └── place_order.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── ports/
│       ├── __init__.py
│       ├── repositories.py
│       └── external_services.py
│
├── infrastructure/         # Externe Schnittstellen
│   ├── __init__.py
│   ├── persistence/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── models.py
│   │   │   └── repositories.py
│   │   └── mongodb/
│   │       └── repositories.py
│   ├── messaging/
│   │   ├── __init__.py
│   │   └── rabbitmq_publisher.py
│   ├── external_services/
│   │   ├── __init__.py
│   │   └── email_service.py
│   └── config.py
│
├── presentation/           # UI/API-Ebene
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── dependencies.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── orders.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user_schemas.py
│   └── cli/
│       └── commands.py
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── main.py                 # Anwendungs-Einstiegspunkt
├── container.py            # Dependency-Injection-Einrichtung
├── pyproject.toml
└── README.md

Testen von Clean Architecture

Unit-Testing von Domain-Logik

import pytest
from uuid import uuid4

def test_user_creation():
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert user.email == "test@example.com"
    assert user.is_active is True

def test_order_total_calculation():
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    order.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert order.total().amount == 25.0

Integrationstests mit Repository

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

def test_user_repository_save_and_retrieve(in_memory_repository):
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )

    saved_user = in_memory_repository.save(user)
    retrieved_user = in_memory_repository.get_by_id(user.id)

    assert retrieved_user is not None
    assert retrieved_user.email == user.email

Testen der Service-Schicht

from unittest.mock import Mock

def test_user_registration():
    # Arrange
    mock_repository = Mock(spec=UserRepository)
    mock_repository.get_by_email.return_value = None
    mock_repository.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )

    mock_email = Mock(spec=EmailSender)
    mock_events = Mock(spec=EventPublisher)

    service = UserService(mock_repository, mock_email, mock_events)

    # Act
    user = service.register_user("test@example.com", "Test")

    # Assert
    assert user.email == "test@example.com"
    mock_repository.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_events.publish.assert_called_once()

Häufige Fallstricke und wie man sie vermeidet

Über-Engineering

Implementieren Sie Clean Architecture nicht für einfache CRUD-Anwendungen. Beginnen Sie einfach und refaktorieren Sie, wenn die Komplexität zunimmt.

Leaky Abstractions

Stellen Sie sicher, dass Domain-Entitäten keine Datenbank-Annotationen oder rahmenwerkspezifischen Code enthalten:

# Schlecht
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Framework sickert in die Domain ein

# Gut
@dataclass
class User:
    id: UUID  # Reine Domain-Entität

Zirkuläre Abhängigkeiten

Verwenden Sie Dependency Injection und Schnittstellen, um zirkuläre Abhängigkeiten zwischen den Ebenen zu vermeiden.

Ignorieren des Kontexts

Clean Architecture ist nicht universell anwendbar. Passen Sie die Strenge der Ebenen basierend auf der Projektgröße und dem Teamwissen an.