Clean Architecture

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

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?

SOLID Principles Checklist

Use this checklist when designing or reviewing code architecture.

Single Responsibility Principle (SRP)

• [ ] Each class has only ONE reason to change
• [ ] Class names clearly describe their single purpose
• [ ] Methods within a class are cohesive (all relate to same responsibility)
• [ ] No "Manager", "Handler", "Processor" suffix (often indicates multiple responsibilities)
• [ ] Services don't mix business logic with infrastructure concerns

Red Flags:

• Class imports from many unrelated modules
• Methods that don't use most class attributes
• Class has > 200 lines (usually)
• Changes to unrelated features require modifying same class

Open/Closed Principle (OCP)

• [ ] New behavior added via new classes, not modifying existing ones
• [ ] Using Protocols/ABCs for extension points
• [ ] Strategy pattern for varying algorithms
• [ ] No switch statements on type (use polymorphism)
• [ ] Configuration over code for variation

Red Flags:

• Growing if/elif chains checking types
• Methods that need modification for each new feature
• Direct instantiation of concrete classes in business logic

Liskov Substitution Principle (LSP)

• [ ] Subclasses don't strengthen preconditions (method requirements)
• [ ] Subclasses don't weaken postconditions (what method guarantees)
• [ ] Subclasses don't throw unexpected exceptions
• [ ] All Protocol methods implemented with compatible signatures
• [ ] Tests pass with any implementation of a Protocol

Red Flags:

• Subclass overrides method to throw NotImplementedError
• Subclass returns different types than base
• Code checks isinstance before calling methods
• Subclass ignores/overrides parent behavior unexpectedly

Interface Segregation Principle (ISP)

• [ ] Protocols are small and focused (3-5 methods max)
• [ ] Clients don't depend on methods they don't use
• [ ] No "god interfaces" with many unrelated methods
• [ ] Role-based interfaces (IReadable, IWritable) vs. object-based
• [ ] Composition of small interfaces over large monolithic ones

Red Flags:

• Implementations that stub out methods with pass or raise
• Protocols with > 10 methods
• Classes implement interface but use only subset of methods
• Interface named after implementation, not capability

Dependency Inversion Principle (DIP)

• [ ] High-level modules don't import from low-level modules
• [ ] Both depend on abstractions (Protocols)
• [ ] Abstractions don't depend on details
• [ ] Dependencies injected, not created internally
• [ ] Domain layer has zero infrastructure imports

Red Flags:

• import from infrastructure in domain/application layer
• Direct instantiation with SomeService() in business logic
• Hardcoded database connections, file paths, URLs
• Tests require actual database/network

Architecture Review Checklist

Layer Independence

• [ ] Domain layer: Zero imports from other layers
• [ ] Application layer: Imports only from Domain
• [ ] Infrastructure layer: Implements ports from Application
• [ ] API layer: Translates DTOs ↔ Domain objects

Dependency Injection

• [ ] All dependencies passed via constructor
• [ ] No global state or singletons in business logic
• [ ] FastAPI Depends() used for wiring
• [ ] Test doubles easily substitutable

Domain Purity

• [ ] Entities use dataclasses, not ORM models
• [ ] No framework imports in domain
• [ ] Value objects are immutable (frozen=True)
• [ ] Domain logic has no side effects (I/O)

Testability

• [ ] Unit tests need no mocks (domain layer)
• [ ] Integration tests mock only external boundaries
• [ ] No database needed for domain logic tests
• [ ] Fast test execution (< 1 second for unit tests)

Quick Reference

When to Refactor

1. Adding feature requires modifying core classes → Extract interface, use OCP 2. Test setup is complex → Apply DIP, inject dependencies 3. Class is growing large → Apply SRP, extract classes 4. Subclass behaves differently → Check LSP, maybe use composition 5. Implementing interface partially → Apply ISP, split interface

FastAPI Clean Architecture Example

Complete example implementing clean architecture patterns in FastAPI.

Project Structure

backend/
├── app/
│   ├── api/v1/
│   │   ├── routes/
│   │   │   └── analyses.py      # Driving adapter
│   │   ├── schemas/
│   │   │   └── analysis.py      # DTOs
│   │   └── deps.py              # DI configuration
│   ├── application/
│   │   ├── services/
│   │   │   └── analysis_service.py
│   │   └── ports/
│   │       └── repositories.py   # Output ports
│   ├── domain/
│   │   ├── entities/
│   │   │   └── analysis.py
│   │   └── value_objects/
│   │       └── analysis_type.py
│   └── infrastructure/
│       └── persistence/
│           ├── models/
│           │   └── analysis_model.py
│           └── repositories/
│               └── postgres_analysis_repo.py
└── tests/
    ├── unit/
    ├── integration/
    └── e2e/

Domain Layer

Entity

# domain/entities/analysis.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from uuid import UUID, uuid4

class AnalysisStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Analysis:
    """Aggregate root for analysis domain."""

    source_url: str
    status: AnalysisStatus = AnalysisStatus.PENDING
    id: UUID = field(default_factory=uuid4)
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    summary: str | None = None
    error_message: str | None = None

    def start_processing(self) -> None:
        if self.status != AnalysisStatus.PENDING:
            raise ValueError(f"Cannot start processing from {self.status}")
        self.status = AnalysisStatus.PROCESSING

    def complete(self, summary: str) -> None:
        if self.status != AnalysisStatus.PROCESSING:
            raise ValueError(f"Cannot complete from {self.status}")
        self.status = AnalysisStatus.COMPLETED
        self.summary = summary

    def fail(self, error: str) -> None:
        self.status = AnalysisStatus.FAILED
        self.error_message = error

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

Value Object

# domain/value_objects/analysis_type.py
from dataclasses import dataclass
from enum import Enum

class ContentType(Enum):
    ARTICLE = "article"
    VIDEO = "video"
    GITHUB_REPO = "github_repo"

@dataclass(frozen=True)
class AnalysisType:
    """Immutable value object for analysis configuration."""

    content_type: ContentType
    depth: int  # 1-3

    def __post_init__(self):
        if not 1 <= self.depth <= 3:
            raise ValueError(f"Depth must be 1-3, got {self.depth}")

    @property
    def is_deep(self) -> bool:
        return self.depth == 3

Application Layer

Output Port (Protocol)

# application/ports/repositories.py
from typing import Protocol
from app.domain.entities.analysis import Analysis

class IAnalysisRepository(Protocol):
    """Output port for analysis persistence."""

    async def save(self, analysis: Analysis) -> Analysis:
        """Persist an analysis."""
        ...

    async def get_by_id(self, id: str) -> Analysis | None:
        """Retrieve analysis by ID."""
        ...

    async def find_by_status(self, status: str) -> list[Analysis]:
        """Find all analyses with given status."""
        ...

Application Service (Use Case)

# application/services/analysis_service.py
from app.application.ports.repositories import IAnalysisRepository
from app.domain.entities.analysis import Analysis

class AnalysisService:
    """Application service implementing use cases."""

    def __init__(self, repo: IAnalysisRepository):
        self._repo = repo

    async def create_analysis(self, url: str) -> Analysis:
        """Create a new analysis."""
        analysis = Analysis(source_url=url)
        return await self._repo.save(analysis)

    async def get_analysis(self, id: str) -> Analysis | None:
        """Get analysis by ID."""
        return await self._repo.get_by_id(id)

    async def start_processing(self, id: str) -> Analysis:
        """Start processing an analysis."""
        analysis = await self._repo.get_by_id(id)
        if not analysis:
            raise ValueError(f"Analysis {id} not found")

        analysis.start_processing()
        return await self._repo.save(analysis)

    async def complete_analysis(self, id: str, summary: str) -> Analysis:
        """Mark analysis as complete."""
        analysis = await self._repo.get_by_id(id)
        if not analysis:
            raise ValueError(f"Analysis {id} not found")

        analysis.complete(summary)
        return await self._repo.save(analysis)

Infrastructure Layer

ORM Model

# infrastructure/persistence/models/analysis_model.py
from sqlalchemy import String, DateTime, Enum as SQLEnum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
import uuid

from app.domain.entities.analysis import Analysis, AnalysisStatus
from app.infrastructure.persistence.base import Base

class AnalysisModel(Base):
    """SQLAlchemy model for analysis."""

    __tablename__ = "analyses"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4
    )
    source_url: Mapped[str] = mapped_column(String(2048))
    status: Mapped[str] = mapped_column(
        SQLEnum(AnalysisStatus, name="analysis_status")
    )
    summary: Mapped[str | None] = mapped_column(String, nullable=True)
    error_message: Mapped[str | None] = mapped_column(String, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        default=lambda: datetime.now(timezone.utc)
    )

    @classmethod
    def from_domain(cls, analysis: Analysis) -> "AnalysisModel":
        """Map domain entity to ORM model."""
        return cls(
            id=analysis.id,
            source_url=analysis.source_url,
            status=analysis.status,
            summary=analysis.summary,
            error_message=analysis.error_message,
            created_at=analysis.created_at,
        )

    def to_domain(self) -> Analysis:
        """Map ORM model to domain entity."""
        return Analysis(
            id=self.id,
            source_url=self.source_url,
            status=AnalysisStatus(self.status),
            summary=self.summary,
            error_message=self.error_message,
            created_at=self.created_at,
        )

Repository Implementation (Driven Adapter)

# infrastructure/persistence/repositories/postgres_analysis_repo.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.application.ports.repositories import IAnalysisRepository
from app.domain.entities.analysis import Analysis, AnalysisStatus
from app.infrastructure.persistence.models.analysis_model import AnalysisModel

class PostgresAnalysisRepository:
    """PostgreSQL implementation of IAnalysisRepository."""

    def __init__(self, session: AsyncSession):
        self._session = session

    async def save(self, analysis: Analysis) -> Analysis:
        model = AnalysisModel.from_domain(analysis)
        self._session.add(model)
        await self._session.flush()
        await self._session.refresh(model)
        return model.to_domain()

    async def get_by_id(self, id: str) -> Analysis | None:
        stmt = select(AnalysisModel).where(AnalysisModel.id == id)
        result = await self._session.execute(stmt)
        model = result.scalar_one_or_none()
        return model.to_domain() if model else None

    async def find_by_status(self, status: str) -> list[Analysis]:
        stmt = select(AnalysisModel).where(
            AnalysisModel.status == AnalysisStatus(status)
        )
        result = await self._session.execute(stmt)
        return [m.to_domain() for m in result.scalars()]

API Layer (Driving Adapter)

Schemas (DTOs)

# api/v1/schemas/analysis.py
from pydantic import BaseModel, HttpUrl, ConfigDict
from datetime import datetime

class CreateAnalysisRequest(BaseModel):
    """Request DTO for creating analysis."""
    url: HttpUrl

class AnalysisResponse(BaseModel):
    """Response DTO for analysis."""
    id: str
    source_url: str
    status: str
    summary: str | None = None
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

    @classmethod
    def from_domain(cls, analysis) -> "AnalysisResponse":
        return cls(
            id=str(analysis.id),
            source_url=analysis.source_url,
            status=analysis.status.value,
            summary=analysis.summary,
            created_at=analysis.created_at,
        )

Dependencies

# api/v1/deps.py
from typing import AsyncGenerator
from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession

from app.application.services.analysis_service import AnalysisService
from app.infrastructure.persistence.repositories.postgres_analysis_repo import (
    PostgresAnalysisRepository
)

async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSession(request.app.state.db_engine) as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

def get_analysis_service(
    db: AsyncSession = Depends(get_db)
) -> AnalysisService:
    repo = PostgresAnalysisRepository(db)
    return AnalysisService(repo)

Routes

# api/v1/routes/analyses.py
from fastapi import APIRouter, Depends, HTTPException, status

from app.api.v1.deps import get_analysis_service
from app.api.v1.schemas.analysis import (
    CreateAnalysisRequest,
    AnalysisResponse
)
from app.application.services.analysis_service import AnalysisService

router = APIRouter(prefix="/analyses", tags=["analyses"])

@router.post("/", response_model=AnalysisResponse, status_code=201)
async def create_analysis(
    request: CreateAnalysisRequest,
    service: AnalysisService = Depends(get_analysis_service),
) -> AnalysisResponse:
    """Create a new analysis."""
    analysis = await service.create_analysis(str(request.url))
    return AnalysisResponse.from_domain(analysis)

@router.get("/{analysis_id}", response_model=AnalysisResponse)
async def get_analysis(
    analysis_id: str,
    service: AnalysisService = Depends(get_analysis_service),
) -> AnalysisResponse:
    """Get analysis by ID."""
    analysis = await service.get_analysis(analysis_id)
    if not analysis:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Analysis {analysis_id} not found"
        )
    return AnalysisResponse.from_domain(analysis)

Tests

Unit Test (Domain)

# tests/unit/domain/test_analysis.py
import pytest
from app.domain.entities.analysis import Analysis, AnalysisStatus

def test_analysis_starts_processing():
    analysis = Analysis(source_url="https://example.com")

    analysis.start_processing()

    assert analysis.status == AnalysisStatus.PROCESSING

def test_analysis_cannot_start_if_not_pending():
    analysis = Analysis(source_url="https://example.com")
    analysis.status = AnalysisStatus.COMPLETED

    with pytest.raises(ValueError):
        analysis.start_processing()

def test_analysis_completes_with_summary():
    analysis = Analysis(source_url="https://example.com")
    analysis.start_processing()

    analysis.complete("Analysis complete")

    assert analysis.status == AnalysisStatus.COMPLETED
    assert analysis.summary == "Analysis complete"

Integration Test (Service)

# tests/integration/test_analysis_service.py
import pytest
from unittest.mock import AsyncMock

from app.application.services.analysis_service import AnalysisService
from app.domain.entities.analysis import Analysis, AnalysisStatus

@pytest.fixture
def mock_repo():
    return AsyncMock()

@pytest.fixture
def service(mock_repo):
    return AnalysisService(repo=mock_repo)

async def test_create_analysis(service, mock_repo):
    mock_repo.save.return_value = Analysis(source_url="https://example.com")

    result = await service.create_analysis("https://example.com")

    assert result.source_url == "https://example.com"
    assert result.status == AnalysisStatus.PENDING
    mock_repo.save.assert_called_once()

Hexagonal Architecture (Ports & Adapters)

Comprehensive guide to implementing hexagonal architecture in Python/FastAPI backends.

Core Concepts

Ports

Interfaces (Python Protocols) that define how the application core communicates with the outside world.

Driving Ports (Primary): How the outside world calls the application

# Input port - what the application offers
class IAnalysisService(Protocol):
    async def create_analysis(self, request: CreateAnalysisRequest) -> Analysis: ...
    async def get_analysis(self, id: str) -> Analysis | None: ...

Driven Ports (Secondary): How the application calls external systems

# Output port - what the application needs
class IAnalysisRepository(Protocol):
    async def save(self, analysis: Analysis) -> Analysis: ...
    async def get_by_id(self, id: str) -> Analysis | None: ...

class INotificationService(Protocol):
    async def send(self, user_id: str, message: str) -> None: ...

Adapters

Concrete implementations that connect ports to external systems.

Driving Adapters (Primary): Translate external requests into application calls

# FastAPI route adapter
@router.post("/analyses")
async def create_analysis(
    request: AnalyzeRequest,
    service: IAnalysisService = Depends(get_analysis_service)
) -> AnalysisResponse:
    analysis = await service.create_analysis(request.to_domain())
    return AnalysisResponse.from_domain(analysis)

Driven Adapters (Secondary): Implement ports using external technologies

# PostgreSQL adapter
class PostgresAnalysisRepository:
    def __init__(self, session: AsyncSession):
        self._session = session

    async def save(self, analysis: Analysis) -> Analysis:
        model = AnalysisModel.from_domain(analysis)
        self._session.add(model)
        await self._session.flush()
        return model.to_domain()

Layer Structure

┌──────────────────────────────────────────────────────────────────┐
│                         INFRASTRUCTURE                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │  FastAPI    │  │  PostgreSQL │  │   Redis     │              │
│  │  Routes     │  │  Repository │  │   Cache     │              │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘              │
│         │                │                │                      │
│         ▼                ▼                ▼                      │
│  ┌──────────────────────────────────────────────────────────────┐│
│  │                     APPLICATION LAYER                        ││
│  │  ┌────────────────────────────────────────────────────────┐  ││
│  │  │              Use Cases / Application Services          │  ││
│  │  │  ┌──────────────────┐  ┌──────────────────────────┐   │  ││
│  │  │  │ AnalysisService  │  │ UserService               │   │  ││
│  │  │  │ - create()       │  │ - register()              │   │  ││
│  │  │  │ - process()      │  │ - authenticate()          │   │  ││
│  │  │  └──────────────────┘  └──────────────────────────┘   │  ││
│  │  └────────────────────────────────────────────────────────┘  ││
│  │                                                              ││
│  │  ┌────────────────────────────────────────────────────────┐  ││
│  │  │                    DOMAIN LAYER                        │  ││
│  │  │  ┌──────────────┐  ┌──────────────┐  ┌─────────────┐  │  ││
│  │  │  │   Entities   │  │ Value Objects│  │   Events    │  │  ││
│  │  │  │   Analysis   │  │ AnalysisType │  │ Completed   │  │  ││
│  │  │  └──────────────┘  └──────────────┘  └─────────────┘  │  ││
│  │  │                                                        │  ││
│  │  │  ┌──────────────────────────────────────────────────┐  │  ││
│  │  │  │              Domain Services                     │  │  ││
│  │  │  │  ScoringService, ValidationService               │  │  ││
│  │  │  └──────────────────────────────────────────────────┘  │  ││
│  │  └────────────────────────────────────────────────────────┘  ││
│  └──────────────────────────────────────────────────────────────┘│
└──────────────────────────────────────────────────────────────────┘

Directory Mapping

backend/app/
├── api/v1/                      # Driving adapters
│   ├── routes/
│   │   ├── analyses.py          # HTTP adapter
│   │   └── users.py
│   ├── schemas/                 # DTOs (request/response)
│   │   ├── analysis.py
│   │   └── user.py
│   └── deps.py                  # Dependency injection
│
├── application/                 # Application layer
│   ├── services/               # Use cases
│   │   ├── analysis_service.py
│   │   └── user_service.py
│   └── ports/                  # Port definitions
│       ├── repositories.py     # Output ports
│       └── services.py         # External service ports
│
├── domain/                      # Domain layer (pure Python)
│   ├── entities/
│   │   ├── analysis.py         # Aggregate root
│   │   └── artifact.py         # Entity
│   ├── value_objects/
│   │   ├── analysis_type.py
│   │   └── url.py
│   ├── events/
│   │   └── analysis_events.py
│   └── services/               # Domain services
│       └── scoring_service.py
│
└── infrastructure/              # Driven adapters
    ├── persistence/
    │   ├── models/             # ORM models
    │   ├── repositories/       # Repository implementations
    │   └── mappers/            # Domain ↔ ORM mappers
    ├── cache/
    │   └── redis_cache.py
    └── external/
        └── llm_client.py

Dependency Rule

Dependencies point inward. Outer layers depend on inner layers, never the reverse.

Infrastructure → Application → Domain
      ↓               ↓           ↓
  (knows)        (knows)    (knows nothing)

Import Rules

# ✅ ALLOWED: Infrastructure imports from Application
# infrastructure/repositories/postgres_analysis_repo.py
from app.application.ports.repositories import IAnalysisRepository
from app.domain.entities.analysis import Analysis

# ✅ ALLOWED: Application imports from Domain
# application/services/analysis_service.py
from app.domain.entities.analysis import Analysis
from app.domain.events import AnalysisCreated

# ❌ FORBIDDEN: Domain imports from Application or Infrastructure
# domain/entities/analysis.py
from app.infrastructure.database import engine  # NEVER!
from app.application.services import something  # NEVER!

Testing Strategy

Unit Tests (Domain Layer)

# No mocks needed - pure Python
def test_analysis_completes():
    analysis = Analysis(id="123", status=AnalysisStatus.PENDING)
    analysis.complete(summary="Done")
    assert analysis.status == AnalysisStatus.COMPLETED

Integration Tests (Application Layer)

# Mock driven ports only
async def test_create_analysis():
    mock_repo = Mock(spec=IAnalysisRepository)
    mock_repo.save.return_value = Analysis(id="123")

    service = AnalysisService(repo=mock_repo)
    result = await service.create(CreateAnalysisRequest(url="..."))

    assert result.id == "123"
    mock_repo.save.assert_called_once()

E2E Tests (Driving Adapters)

# Full stack with test database
async def test_create_analysis_endpoint(client: TestClient, db: AsyncSession):
    response = await client.post("/api/v1/analyses", json={"url": "..."})
    assert response.status_code == 201

Related Files

• See checklists/solid-checklist.md for SOLID principles checklist
• See scripts/domain-entity-template.py for entity templates
• See SKILL.md for DDD patterns