Clean Architecture Patterns

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

SOLID Principles ( 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 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

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