Clean Architecture Patterns
Build maintainable, testable backends with SOLID principles and hexagonal architecture.
SOLID Principles (2026 Python)
S - Single Responsibility
# BAD: One class doing everything
class UserManager:
def create_user(self, data): ...
def send_welcome_email(self, user): ...
def generate_report(self, users): ...
# GOOD: Separate responsibilities
class UserService:
def create_user(self, data: UserCreate) -> User: ...
class EmailService:
def send_welcome(self, user: User) -> None: ...
class ReportService:
def generate_user_report(self, users: list[User]) -> Report: ...O - Open/Closed (Protocol-based)
from typing import Protocol
class PaymentProcessor(Protocol):
async def process(self, amount: Decimal) -> PaymentResult: ...
class StripeProcessor:
async def process(self, amount: Decimal) -> PaymentResult:
# Stripe implementation
...
class PayPalProcessor:
async def process(self, amount: Decimal) -> PaymentResult:
# PayPal implementation - extends without modifying
...L - Liskov Substitution
# Any implementation of Repository can substitute another
class IUserRepository(Protocol):
async def get_by_id(self, id: str) -> User | None: ...
async def save(self, user: User) -> User: ...
class PostgresUserRepository:
async def get_by_id(self, id: str) -> User | None: ...
async def save(self, user: User) -> User: ...
class InMemoryUserRepository: # For testing - fully substitutable
async def get_by_id(self, id: str) -> User | None: ...
async def save(self, user: User) -> User: ...I - Interface Segregation
# BAD: Fat interface
class IRepository(Protocol):
async def get(self, id: str): ...
async def save(self, entity): ...
async def delete(self, id: str): ...
async def search(self, query: str): ...
async def bulk_insert(self, entities): ...
# GOOD: Segregated interfaces
class IReader(Protocol):
async def get(self, id: str) -> T | None: ...
class IWriter(Protocol):
async def save(self, entity: T) -> T: ...
class ISearchable(Protocol):
async def search(self, query: str) -> list[T]: ...D - Dependency Inversion
from typing import Protocol
from fastapi import Depends
class IAnalysisRepository(Protocol):
async def get_by_id(self, id: str) -> Analysis | None: ...
class AnalysisService:
def __init__(self, repo: IAnalysisRepository):
self._repo = repo # Depends on abstraction, not concrete
# FastAPI DI
def get_analysis_service(
db: AsyncSession = Depends(get_db)
) -> AnalysisService:
repo = PostgresAnalysisRepository(db)
return AnalysisService(repo)Hexagonal Architecture (Ports & Adapters)
┌─────────────────────────────────────────────────────────────┐
│ DRIVING ADAPTERS │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ FastAPI │ │ CLI │ │ Celery │ │ Tests │ │
│ │ Routes │ │ Commands │ │ Tasks │ │ Mocks │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ╔═══════════════════════════════════════════════════════╗ │
│ ║ INPUT PORTS ║ │
│ ║ ┌─────────────────┐ ┌─────────────────────────────┐ ║ │
│ ║ │ AnalysisService │ │ UserService │ ║ │
│ ║ │ (Use Cases) │ │ (Use Cases) │ ║ │
│ ║ └────────┬────────┘ └──────────────┬──────────────┘ ║ │
│ ╠═══════════╪══════════════════════════╪════════════════╣ │
│ ║ ▼ DOMAIN ▼ ║ │
│ ║ ┌─────────────────────────────────────────────────┐ ║ │
│ ║ │ Entities │ Value Objects │ Domain Events │ ║ │
│ ║ │ Analysis │ AnalysisType │ AnalysisCreated │ ║ │
│ ║ └─────────────────────────────────────────────────┘ ║ │
│ ╠═══════════════════════════════════════════════════════╣ │
│ ║ OUTPUT PORTS ║ │
│ ║ ┌──────────────────┐ ┌────────────────────────────┐ ║ │
│ ║ │ IAnalysisRepo │ │ INotificationService │ ║ │
│ ║ │ (Protocol) │ │ (Protocol) │ ║ │
│ ║ └────────┬─────────┘ └──────────────┬─────────────┘ ║ │
│ ╚═══════════╪══════════════════════════╪════════════════╝ │
│ ▼ ▼ │
│ ┌───────────────────┐ ┌────────────────────────────────┐ │
│ │ PostgresRepo │ │ EmailNotificationService │ │
│ │ (SQLAlchemy) │ │ (SMTP/SendGrid) │ │
│ └───────────────────┘ └────────────────────────────────┘ │
│ DRIVEN ADAPTERS │
└─────────────────────────────────────────────────────────────┘DDD Tactical Patterns
Entity (Identity-based)
from dataclasses import dataclass, field
from uuid import UUID, uuid4
@dataclass
class Analysis:
id: UUID = field(default_factory=uuid4)
source_url: str
status: AnalysisStatus
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
def __eq__(self, other: object) -> bool:
if not isinstance(other, Analysis):
return False
return self.id == other.id # Identity equalityValue Object (Structural equality)
from dataclasses import dataclass
@dataclass(frozen=True) # Immutable
class AnalysisType:
category: str
depth: int
def __post_init__(self):
if self.depth < 1 or self.depth > 3:
raise ValueError("Depth must be 1-3")Aggregate Root
class AnalysisAggregate:
def __init__(self, analysis: Analysis, artifacts: list[Artifact]):
self._analysis = analysis
self._artifacts = artifacts
self._events: list[DomainEvent] = []
def complete(self, summary: str) -> None:
self._analysis.status = AnalysisStatus.COMPLETED
self._analysis.summary = summary
self._events.append(AnalysisCompleted(self._analysis.id))
def collect_events(self) -> list[DomainEvent]:
events = self._events.copy()
self._events.clear()
return eventsDirectory Structure
backend/app/
├── api/v1/ # Driving adapters (FastAPI routes)
├── domains/
│ └── analysis/
│ ├── entities.py # Domain entities
│ ├── value_objects.py # Value objects
│ ├── services.py # Domain services (use cases)
│ ├── repositories.py # Output port protocols
│ └── events.py # Domain events
├── infrastructure/
│ ├── repositories/ # Driven adapters (PostgreSQL)
│ ├── services/ # External service adapters
│ └── messaging/ # Event publishers
└── core/
├── dependencies.py # FastAPI DI configuration
└── protocols.py # Shared protocolsAnti-Patterns (FORBIDDEN)
# NEVER import infrastructure in domain
from app.infrastructure.database import engine # In domain layer
# NEVER leak ORM models to API
@router.get("/users/{id}")
async def get_user(id: str, db: Session) -> UserModel: # Returns ORM model
return db.query(UserModel).get(id)
# NEVER have domain depend on framework
from fastapi import HTTPException
class UserService:
def get(self, id: str):
if not user:
raise HTTPException(404) # Framework in domain!Key Decisions
| Decision | Recommendation |
|---|---|
| Protocol vs ABC | Use Protocol (structural typing) |
| Dataclass vs Pydantic | Dataclass for domain, Pydantic for API |
| Repository granularity | One per aggregate root |
| Transaction boundary | Service layer, not repository |
| Event publishing | Collect in aggregate, publish after commit |
Related Skills
repository-patterns- Detailed repository implementationsapi-design-framework- REST API patternsdatabase-schema-designer- Schema design
Capability Details
solid-principles
Keywords: SOLID, single responsibility, open closed, liskov, interface segregation, dependency inversion Solves:
- How do I apply SOLID principles in Python?
- My classes are doing too much
hexagonal-architecture
Keywords: hexagonal, ports and adapters, clean architecture, onion Solves:
- How do I structure my FastAPI app?
- How to separate infrastructure from domain?
ddd-tactical
Keywords: entity, value object, aggregate, domain event, DDD Solves:
- What's the difference between entity and value object?
- How to design aggregates?