Python Design Patterns voor Clean Architecture
Maak onderhoudbare Python-apps met SOLID-ontwerppatronen
Clean Architecture heeft de manier waarop ontwikkelaars schaalbare, onderhoudbare toepassingen bouwen, veranderd door nadruk te leggen op het scheiden van zorgen en het beheren van afhankelijkheden.
In Python combineren deze principes de dynamische aard van de taal tot flexibele, testbare systemen die evolueren met de businessbehoeften zonder technische schulden te worden.

Begrijpen van Clean Architecture in Python
Clean Architecture, geïntroduceerd door Robert C. Martin (Uncle Bob), organiseert software in concentrische lagen waarin afhankelijkheden naar binnen wijzen naar kernbedrijfslogica. Dit architecturale patroon zorgt ervoor dat de kritieke bedrijfsregels van uw toepassing onafhankelijk zijn van frameworks, databases en externe diensten.
De Kernfilosofie
Het fundamentele principe is eenvoudig maar krachtig: bedrijfslogica mag niet afhankelijk zijn van infrastructuur. Uw domeinobjecten, use cases en bedrijfsregels moeten werken ongeacht of u PostgreSQL of MongoDB, FastAPI of Flask, AWS of Azure gebruikt.
In Python past deze filosofie perfect bij de taal’s “duck typing” en protocolgeoriënteerde programmering, waardoor een duidelijke scheiding mogelijk is zonder de ceremonie die statisch getypeerde talen vereisen.
De Vier Lagen van Clean Architecture
Entiteitenlaag (Domein): Pure bedrijfsobjecten met bedrijfsregels die van toepassing zijn op het hele bedrijf. Deze zijn POJOs (Plain Old Python Objects) zonder externe afhankelijkheden.
Use Cases Laag (Toepassing): Toepassingsgerichte bedrijfsregels die de stroom van data tussen entiteiten en externe diensten coördineren.
Interface Adapters Laag: Converteert data tussen het formaat dat het meest geschikt is voor use cases en entiteiten, en het formaat dat door externe partijen vereist wordt.
Frameworks & Drivers Laag: Alle externe details zoals databases, webframeworks en externe APIs.
SOLID Principes in Python
De SOLID principes vormen de basis van clean architecture. Laten we kijken hoe elk principe zich in Python manifesteert. Voor een uitgebreid overzicht van design patterns in Python, zie de Python Design Patterns Guide.
Single Responsibility Principle (SRP)
Elke klasse moet slechts één reden hebben om te veranderen:
# Slecht: Meerdere verantwoordelijkheden
class UserManager:
def create_user(self, user_data):
# Creëer gebruiker
pass
def send_welcome_email(self, user):
# Stuur e-mail
pass
def log_creation(self, user):
# Log naar bestand
pass
# Goed: Verantwoordelijkheden gescheiden
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"Gebruiker aangemaakt: {user.id}")
return user
Open/Closed Principle (OCP)
Softwareentiteiten moeten open staan voor uitbreiding, maar gesloten voor wijziging:
from abc import ABC, abstractmethod
from typing import Protocol
# Gebruik Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# Credit card logica
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# PayPal logica
return True
# Gemakkelijk uitbreidbaar zonder bestaande code te wijzigen
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# Cryptocurrency logica
return True
Liskov Substitution Principle (LSP)
Objecten moeten vervangen kunnen worden door hun subtypes zonder het programma te breken:
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 implementatie
pass
def get(self, key: str) -> str:
# PostgreSQL implementatie
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Redis implementatie
pass
def get(self, key: str) -> str:
# Redis implementatie
return ""
# Beide kunnen wisselend gebruikt worden
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
Interface Segregation Principle (ISP)
Clients mogen niet gedwongen worden om afhankelijk te zijn van interfaces die ze niet gebruiken:
# Slecht: Vette interface
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# Goed: Gesegmenteerde interfaces
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("Werken")
def eat(self) -> None:
print("Eten")
class Robot:
def work(self) -> None:
print("Werken")
# Geen eat methode nodig
Dependency Inversion Principle (DIP)
Hoog niveau modules mogen niet afhankelijk zijn van laag niveau modules. Beide moeten afhankelijk zijn van abstracties:
from typing import Protocol
# Abstractie
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# Laag niveau module
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# SMTP implementatie
pass
# Hoog niveau module afhankelijk van abstractie
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# Registratie logica
self.email_sender.send(
to=email,
subject="Welkom!",
body=f"Hallo {name}"
)
Repository Pattern: Abstracteren van Data Access
Het Repository Pattern biedt een collectie-achtige interface voor het toegang krijgen tot domeinobjecten, waarbij de details van gegevensopslag worden verborgen.
Basis Repository Implementatie
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 Implementatie
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 voor Testen
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 Laag: Orchestration van Bedrijfslogica
De Service Laag implementeert use cases en orchestreert de stroom tussen repositories, externe diensten en domeinlogica.
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:
# Controleer of gebruiker bestaat
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"Gebruiker met e-mail {email} bestaat al")
# Nieuwe gebruiker aanmaken
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# Opslaan in repository
user = self.user_repository.save(user)
# Welkomse-mail sturen
self.email_service.send(
to=user.email,
subject="Welkom!",
body=f"Hallo {user.name}, welkom op onze platform!"
)
# Gebeurtenis publiceren
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"Gebruiker {user_id} niet gevonden")
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
De dynamische aard van Python maakt dependency injection eenvoudig zonder zware frameworks te vereisen.
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):
# Gebruik ingevoegde afhankelijkheden
pass
Eenvoudige 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"Geen registratie gevonden voor {interface}")
# Gebruik
def create_container() -> Container:
container = Container()
# Services registreren
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 Architectuur (Poorten en Adapters)
Hexagonale Architectuur plaatst bedrijfslogica in het midden met adapters die externe communicatie afhandelen.
Definiëren van Poorten (Interfaces)
# Invoerpoort (Primair)
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# Uitvoerpoort (Secundair)
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
Implementeren van Adapters
from pydantic import BaseModel, EmailStr
# Invoeradapter (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))
# Uitvoeradapter (Database)
# Al geïmplementeerd als SQLAlchemyUserRepository
Domain-Driven Design Patterns
Value Objects
Onveranderlijke objecten gedefinieerd door hun attributen:
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"Ongeldige 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("Bedrag mag niet negatief zijn")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"Onondersteunde valuta: {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Kan verschillende valuta's niet optellen")
return Money(self.amount + other.amount, self.currency)
Aggregaten
Groepeer van domeinobjecten behandeld als een enkel eenheid:
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("Kan lege bestelling niet bevestigen")
if self.status != "pending":
raise ValueError("Bestelling al verwerkt")
self.status = "confirmed"
Domain Events
Domain events bevorderen loskoppeling tussen componenten en ondersteunen event-driven architectuur. Voor productie-georiënteerde event-driven systemen, overweeg het implementeren van event streaming met diensten zoals AWS Kinesis—zie Building Event-Driven Microservices with AWS Kinesis voor een gedetailleerde gids.
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 functies voor een schone architectuur
De moderne functies van Python maken het implementeren van een schone architectuur eleganter en typeveiler. Als je een snelle verwijzing nodig hebt voor Python syntaxis en functies, bekijk dan de Python Cheatsheet.
Typehints en Protocols
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 voor validatie
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('Naam mag geen cijfers bevatten')
return v
class Config:
frozen = True # Maak onveranderlijk
Async/Await voor I/O-acties
De async/await syntaxis van Python is vooral krachtig voor I/O-gebonden acties in een schone architectuur, wat niet-blokkerende interacties met databases en externe diensten mogelijk maakt. Bij het implementeren van Python-applicaties op serverless platforms wordt het begrip van prestatiekenmerken cruciaal—zie AWS lambda prestaties: JavaScript vs Python vs Golang voor inzichten over het optimaliseren van Python-serverless functies.
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]
Beste praktijken voor projectstructuur
Een goede projectorganisatie is essentieel voor het behouden van een schone architectuur. Voordat je je projectstructuur instelt, zorg er dan voor dat je Python virtuele omgevingen gebruikt voor afhankelijkheid isolatie. De venv Cheatsheet behandelt alles wat je moet weten over het beheren van virtuele omgevingen. Voor moderne Python projecten, overweeg dan uv - Nieuwe Python pakket, project en omgevingsbeheerder, die snellere pakketbeheer en projectopzet biedt.
my_application/
├── domain/ # Ondernemingsbedrijfsregels
│ ├── __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/ # Toepassingsbedrijfsregels
│ ├── __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 interfaces
│ ├── __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-laag
│ ├── __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 # Toepassingsstartpunt
├── container.py # Inrichting van afhankelijkheden
├── pyproject.toml
└── README.md
Testen van schone architectuur
Eenheidstesten van domeinlogica
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
Integratietesten met 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 van service-laag
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()
Algemene valkuilen en hoe je deze kunt vermijden
Over-engineering
Implementeer geen schone architectuur voor eenvoudige CRUD-toepassingen. Begin eenvoudig en refactoreer als de complexiteit toeneemt.
Leaky abstractions
Zorg ervoor dat domeinobjecten geen databaseannotaties of frameworkspecifieke code bevatten:
# Slecht
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # Framework lekt naar domein
# Goed
@dataclass
class User:
id: UUID # Pura domeinobject
Circulaire afhankelijkheden
Gebruik afhankelijkheidsinjektie en interfaces om circulaire afhankelijkheden tussen lagen te doorbreken.
Genegeerde context
Schone architectuur is niet voor iedereen hetzelfde. Pas de strikteheid van lagen aan op basis van projectgrootte en teamexpertise.
Nuttige links
- Schone architectuur door Robert C. Martin
- Python Type Hints Documentatie
- Pydantic Documentatie
- FastAPI Officiële Documentatie
- SQLAlchemy ORM Documentatie
- Dependency Injector Bibliothek
- Domain-Driven Design Referentie
- Architectuurpatronen met Python
- Martin Fowlers Blog over architectuur
- Python Design Patterns Gids
- Python Cheatsheet
- venv Cheatsheet
- uv - Nieuwe Python pakket, project en omgevingsbeheerder
- AWS lambda prestaties: JavaScript vs Python vs Golang
- Bouwen van gebeurtenisgestuurde microservices met AWS Kinesis