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.
