Clean Architecture Patterns

Build maintainable, testable backends with SOLID principles and hexagonal architecture.

When to Use

• Designing new service layer architecture
• Refactoring tightly-coupled code
• Implementing domain-driven design
• Creating testable business logic
• Separating infrastructure from domain

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=datetime.utcnow)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Analysis):
            return False
        return self.id == other.id  # Identity equality

Value 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 events

Directory 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 protocols

Anti-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

Related Skills

• repository-patterns - Detailed repository implementations
• api-design-framework - REST API patterns
• database-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?