Skip to content

Latest commit

 

History

History
358 lines (281 loc) · 11.9 KB

File metadata and controls

358 lines (281 loc) · 11.9 KB

Clean Architecture Overview

What is Clean Architecture?

Clean Architecture is a software design philosophy that emphasizes:

  • Independence of Frameworks: Business logic is not coupled to frameworks or libraries
  • Testability: Core business logic can be tested without UI, database, or external services
  • Independence of UI: The UI can change without affecting business logic
  • Independence of Database: Database choice is an implementation detail
  • Independence of External Agencies: Business rules don't depend on external APIs or services

The Dependency Rule

Source code dependencies must point inward. Nothing in an inner circle can know anything about outer circles.

This means:

  • ✅ Outer layers can depend on inner layers
  • ❌ Inner layers MUST NOT depend on outer layers
  • ❌ No circular dependencies

Architecture Layers in This Blueprint

1. Domain Layer (Innermost)

Location: src/domain/

The core of the application, containing pure business logic with no external dependencies.

Components:

  • Entities: Core business objects representing concepts in your domain

    • Example: User, Product, Order
    • Contains business rules and invariants
    • Should be framework-agnostic
  • Value Objects: Immutable objects defined by their values, not identity

    • Example: Email, Money, Address
    • Encapsulate validation logic
  • Repository Interfaces: Abstract contracts for data access

    • Define WHAT data operations are needed (not HOW)
    • Enable testing with mock repositories
    • Allow switching database implementations
  • Domain Exceptions: Custom exceptions for business rule violations

    • Example: UserAlreadyExistsError, InsufficientFundsError

Characteristics:

  • ✅ Zero external dependencies
  • ✅ Pure Python code
  • ✅ Highly testable
  • ✅ Reusable in different contexts (CLI, web, batch jobs)

Example:

# Domain layer - pure business logic
class User:
    def __init__(self, id: int, email: str, name: str):
        if not self._is_valid_email(email):
            raise InvalidEmailError()
        self.id = id
        self.email = email
        self.name = name
    
    @staticmethod
    def _is_valid_email(email: str) -> bool:
        return "@" in email

2. Application Layer

Location: src/application/

Orchestrates domain objects and implements use cases. It's the "glue" between domain logic and external concerns.

Components:

  • Use Cases / Application Services: Define application-specific business rules

    • Example: CreateUserUseCase, UpdateProductPriceUseCase
    • Orchestrate domain objects
    • Coordinate with repositories
    • Handle transactions and domain events
  • DTOs (Data Transfer Objects): Simplified data structures for transferring between layers

    • Input DTOs: Carry data into use cases
    • Output DTOs: Carry results out of use cases
  • Mappers: Convert between domain entities and DTOs

    • Decouple internal representation from external communication

Characteristics:

  • ✅ Depends only on domain layer
  • ✅ No knowledge of frameworks
  • ✅ No direct database access (uses repository interfaces)
  • ✅ Handles use case orchestration
  • ✅ Highly testable with mock repositories

Example:

# Application layer - use case orchestration
class CreateUserUseCase:
    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository
    
    async def execute(self, email: str, name: str) -> UserOutputDTO:
        # Check business rule: email uniqueness
        existing_user = await self.user_repository.find_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(email)
        
        # Create domain entity
        user = User(id=None, email=email, name=name)
        
        # Persist through repository interface
        created_user = await self.user_repository.save(user)
        
        # Return DTO
        return UserOutputDTO.from_entity(created_user)

3. Infrastructure Layer

Location: src/infrastructure/

Implements the concrete details of data access, external services, and other technical concerns.

Components:

  • Repository Implementations: Concrete implementations of repository interfaces

    • Example: SQLAlchemyUserRepository, MongoDBUserRepository
    • Handle database queries and persistence
    • Manage ORM/ODM models
  • Database Models: ORM/ODM representations of domain entities

    • Example: SQLAlchemy models
    • Separate from domain entities
    • Framework-specific concerns
  • External Service Adapters: Integrate with external APIs

    • Example: EmailServiceAdapter, PaymentGatewayAdapter
    • Translate external APIs to domain concepts
  • Configuration: Database connections, API clients

    • Initialization of dependencies

Characteristics:

  • ✅ Implements repository interfaces from domain
  • ✅ Handles database-specific logic
  • ✅ Contains framework-specific code (SQLAlchemy, ORMs)
  • ✅ Can have external dependencies
  • ✅ Easily replaceable implementations

Example:

# Infrastructure layer - concrete database implementation
class SQLAlchemyUserRepository(IUserRepository):
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def find_by_email(self, email: str) -> User | None:
        # Database-specific query
        result = await self.session.execute(
            select(UserModel).where(UserModel.email == email)
        )
        user_model = result.scalar_one_or_none()
        
        if not user_model:
            return None
        
        # Map ORM model back to domain entity
        return self._to_domain(user_model)
    
    async def save(self, user: User) -> User:
        # Convert domain entity to ORM model
        user_model = self._to_model(user)
        self.session.add(user_model)
        await self.session.commit()
        return self._to_domain(user_model)
    
    def _to_domain(self, model: UserModel) -> User:
        return User(id=model.id, email=model.email, name=model.name)
    
    def _to_model(self, user: User) -> UserModel:
        return UserModel(id=user.id, email=user.email, name=user.name)

4. Presentation Layer (Outermost)

Location: src/presentation/

Handles HTTP communication, request validation, and API formatting. FastAPI specific code.

Components:

  • FastAPI Routers: Define HTTP endpoints

    • Handle HTTP methods (GET, POST, PUT, DELETE)
    • Define URL paths
    • Specify response codes and descriptions
  • Request/Response Schemas: Pydantic models for validation

    • Validate incoming request data
    • Serialize outgoing response data
    • Document API in OpenAPI
  • Dependencies: FastAPI dependency injection

    • Provide repositories and use cases to routers
    • Handle database sessions
    • Manage transient vs singleton lifecycles
  • Middleware: Cross-cutting concerns

    • Error handling
    • Logging
    • CORS, authentication, authorization

Characteristics:

  • ✅ FastAPI-specific code
  • ✅ Depends on application and domain layers
  • ✅ Handles HTTP protocol details
  • ✅ Validates and formats data

Example:

# Presentation layer - FastAPI routers
from fastapi import APIRouter, Depends, HTTPException

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

async def get_create_user_use_case(
    db_session: AsyncSession = Depends(get_db_session)
) -> CreateUserUseCase:
    repository = SQLAlchemyUserRepository(db_session)
    return CreateUserUseCase(repository)

@router.post("/", status_code=201)
async def create_user(
    request: CreateUserRequest,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case)
) -> UserResponse:
    try:
        result = await use_case.execute(
            email=request.email,
            name=request.name
        )
        return UserResponse(**result.dict())
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=409, detail=str(e))

Data Flow Example: Creating a User

1. HTTP Request arrives at FastAPI router
   POST /api/users { "email": "user@example.com", "name": "John" }
                      ↓
2. Request Schema validates the input (Pydantic)
                      ↓
3. Router receives dependencies (repository, use case)
                      ↓
4. Use Case is executed (application layer)
   - Checks business rule: unique email
   - Creates User entity (domain layer)
   - Calls repository to persist
                      ↓
5. Repository saves to database (infrastructure layer)
   - Converts User entity to ORM model
   - Executes SQL query
   - Returns persisted entity
                      ↓
6. Use Case maps result to DTO
                      ↓
7. Router converts DTO to Response Schema
                      ↓
8. FastAPI serializes to JSON and returns HTTP 201
   { "id": 1, "email": "user@example.com", "name": "John" }

Design Patterns Used

1. Repository Pattern

Abstracts data access, allowing domain logic to remain agnostic about persistence implementation.

# Domain defines interface
class IUserRepository(ABC):
    async def find_by_email(self, email: str) -> User | None: ...
    async def save(self, user: User) -> User: ...

# Infrastructure provides implementation
class SQLAlchemyUserRepository(IUserRepository):
    async def find_by_email(self, email: str) -> User | None: ...

2. Dependency Injection

Loose coupling between components. Dependencies are injected rather than created internally.

# Use case doesn't create repository; it's injected
class CreateUserUseCase:
    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository  # Injected, not created

3. Data Transfer Object (DTO)

Simplifies data transfer between layers, preventing domain model leakage.

# Application layer returns DTO, not domain entity
class UserOutputDTO:
    id: int
    email: str
    name: str

4. Mapper

Converts between domain entities and other representations.

class UserMapper:
    @staticmethod
    def to_domain(user_model: UserModel) -> User: ...
    
    @staticmethod
    def to_response(user: User) -> UserResponse: ...

Key Principles

✅ Single Responsibility Principle (SRP)

Each class has one reason to change. Use cases orchestrate, repositories persist, entities encapsulate rules.

✅ Open/Closed Principle (OCP)

Open for extension (add new repositories), closed for modification (don't change existing domain rules).

✅ Liskov Substitution Principle (LSP)

Repository implementations are interchangeable. You can swap SQLAlchemy for MongoDB without affecting use cases.

✅ Interface Segregation Principle (ISP)

Narrow, focused interfaces. IUserRepository doesn't include product-related methods.

✅ Dependency Inversion Principle (DIP)

Depend on abstractions, not concretions. Use cases depend on IUserRepository, not SQLAlchemyUserRepository.

Advantages of This Architecture

Testability: Domain logic and use cases tested without database ✅ Maintainability: Clear structure, easy to find where to make changes ✅ Flexibility: Swap databases, frameworks without touching business logic ✅ Scalability: Easy to add new features without affecting existing code ✅ Reusability: Domain logic can be used in CLI, scheduled jobs, batch processing ✅ Team Collaboration: Clear boundaries allow developers to work independently ✅ Onboarding: New developers understand the structure quickly

Trade-offs to Consider

⚠️ Complexity: More files and layers for simple CRUD operations ⚠️ Learning Curve: Understanding Clean Architecture principles takes time ⚠️ Initial Overhead: More boilerplate code upfront (pays off in maintainability)

For very small prototypes or scripts, this architecture may be overkill. Use judgment based on project scope.


Further Reading:

  • "Clean Architecture" by Robert C. Martin
  • Domain-Driven Design patterns
  • SOLID principles documentation