Berkay Sonel

Posted on Jun 13

Clean Architecture and Domain-Driven Design (DDD) in FastAPI

#fastapi #ddd #cleancode #architecture

Setting up a scalable backend application requires a clear separation of concerns. While FastAPI allows for rapid prototyping adnd flexibility, applying structured patterns like Clean Architecture and Domain-Driven Design (DDD) can help maintainability as the project grows.

Below is a structured approach to implementing these architectural patterns in a FastAPI application.

Architectural Directory Structure

To separate core business logic from external frameworks, databases, and delivery mechanisms, the codebase can be organized into distinct layers:

backend/
├── domain/         # Core business entities and abstract interfaces (Zero external dependencies)
├── application/    # Application logic and use cases
├── infrastructure/ # Database implementations, file storage, external APIs
└── api/            # Presentation layer (FastAPI routers, dependency injection, schemas)

Implementing the Layers

1. The Domain Layer (Core Business Rules)

The domain layer represents the core of the application. It remains completely isolated from external frameworks, database engines, or third-party libraries. It defines domain entities, value objects, and repository interfaces (ports).

Entity Placeholder

# domain/entities/document.py

class Document:
    """Represents a core business domain entity."""
    def __init__(self, id: str, status: str):
        self.id = id
        self.status = status

    def process(self) -> None:
        """Domain-specific business logic."""
        self.status = "PROCESSING"

Abstract Interface (Port) Placeholder

# domain/interfaces/repository.py
from abc import ABC, abstractmethod

class IDocumentRepository(ABC):
    """Abstract interface defining database capabilities."""

    @abstractmethod
    def save(self, document: Document) -> None:
        pass

    @abstractmethod
    def find_by_id(self, id: str) -> Document:
        pass

2. The Application Layer (Use Cases)

The application layer coordinates the execution flow. It implements specific use cases by depending on the abstract interfaces defined in the domain layer, rather than concrete database or infrastructure classes.

Use Case Placeholder

# application/use_cases/process_document.py
from domain.interfaces.repository import IDocumentRepository

class ProcessDocumentUseCase:
    """Encapsulates a specific application workflow."""

    def __init__(self, repository: IDocumentRepository):
        # Dependencies are injected via the constructor
        self.repository = repository

    def execute(self, document_id: str) -> None:
        # Retrieve domain entity through the abstraction
        document = self.repository.find_by_id(document_id)

        # Execute domain business logic
        document.process()

        # Save updated domain state
        self.repository.save(document)

3. The Infrastructure Layer (Adapters)

The infrastructure layer contains the concrete implementations of the abstract ports defined in the domain. This is where database engines (SQLAlchemy, Tortoise, etc.), file storage systems, and external service clients are integrated.

Repository Adapter Placeholder

# infrastructure/database/sqlite_repository.py
from domain.entities.document import Document
from domain.interfaces.repository import IDocumentRepository

class SqliteDocumentRepository(IDocumentRepository):
    """Concrete SQLite implementation of the repository interface."""

    def save(self, document: Document) -> None:
        # Framework-specific database write logic (e.g., SQLAlchemy / raw SQL)
        pass

    def find_by_id(self, id: str) -> Document:
        # Framework-specific database read logic
        pass

4. The Presentation Layer (FastAPI Routers & Dependency Injection)

In this setup, FastAPI acts strictly as a delivery mechanism. The API layer handles routing, serialization (Pydantic), HTTP status codes, and Dependency Injection (DI) to wire concrete infrastructure adapters into the application use cases.

Dependency Configuration & Router Placeholder

# api/dependencies.py
from fastapi import Depends
from infrastructure.database.sqlite_repository import SqliteDocumentRepository
from application.use_cases.process_document import ProcessDocumentUseCase

# Singleton or session factory definitions
_repository_instance = SqliteDocumentRepository()

def get_repository() -> SqliteDocumentRepository:
    return _repository_instance

def get_process_document_use_case(
    repository: SqliteDocumentRepository = Depends(get_repository)
) -> ProcessDocumentUseCase:
    return ProcessDocumentUseCase(repository)

# api/routes.py
from fastapi import APIRouter, Depends, status
from api.dependencies import get_process_document_use_case
from application.use_cases.process_document import ProcessDocumentUseCase

router = APIRouter()

@router.post("/documents/{document_id}/process", status_code=status.HTTP_200_OK)
def process_document(
    document_id: str,
    use_case: ProcessDocumentUseCase = Depends(get_process_document_use_case)
):
    use_case.execute(document_id)
    return {"message": "Processing started"}

Architectural Advantages

Database & Framework Independence: Core business logic is decoupled from external tools. Transitioning from SQLite to PostgreSQL, or replacing a queue library, requires writing a new adapter rather than modifying the core application rules.
Testability: Business rules can be verified in isolation. Because use cases depend on interfaces, unit tests can run quickly using mock adapters without requiring database connections.
Separation of Concerns: Developers can work on API endpoints, database queries, and business rules within clearly defined boundaries, reducing the risk of unintended side effects in unrelated modules.

DEV Community