Clean Architecture Structure (FastAPI)

Approaching modular projects in Python FastAPI with clean architecture in mind is an excellent way to build scalable, maintainable, and testable applications. Here’s a suggested approach that combines FastAPI’s strengths with clean architecture principles:

project_root/
│
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── exceptions.py
│   │
│   ├── domain/
│   │   ├── __init__.py
│   │   ├── models/
│   │   │   ├── __init__.py
│   │   │   └── user.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user.py
│   │
│   ├── usecases/
│   │   ├── __init__.py
│   │   └── user.py
│   │
│   ├── interfaces/
│   │   ├── __init__.py
│   │   └── repositories/
│   │       ├── __init__.py
│   │       └── user_repository.py
│   │
│   ├── infrastructure/
│   │   ├── __init__.py
│   │   ├── database.py
│   │   └── repositories/
│   │       ├── __init__.py
│   │       └── sqlalchemy_user_repository.py
│   │
│   └── api/
│       ├── __init__.py
│       ├── dependencies.py
│       └── v1/
│           ├── __init__.py
│           └── endpoints/
│               ├── __init__.py
│               └── user.py
│
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api/
│   │   └── test_user_api.py
│   └── test_usecases/
│       └── test_user_usecases.py
│
├── alembic/
│   ├── versions/
│   └── env.py
│
├── .env
├── .gitignore
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Now, let’s break down this structure and discuss how to implement clean architecture principles:

Domain Layer:
- Located in app/domain/
- Contains enterprise business rules and entities
- Includes models/ for ORM models and schemas/ for Pydantic models
Use Case Layer:
- Located in app/usecases/
- Implements application-specific business rules
- Orchestrates the flow of data to and from entities
Interface Adapters Layer:
- Includes app/interfaces/ and app/api/
- interfaces/ defines abstract base classes for repositories
- api/ contains FastAPI routes and request/response models
Infrastructure Layer:
- Located in app/infrastructure/
- Implements interfaces defined in the Interface Adapters layer
- Includes database configurations and concrete repository implementations
Main Application:
- app/main.py is the entry point of the application
- Sets up FastAPI app, includes routers, and configures middleware

Here’s how to implement some key components:

Domain Model:

from sqlalchemy import Column, Integer, String
from app.infrastructure.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

Use Case:

from app.domain.models.user import User
from app.interfaces.repositories.user_repository import UserRepository

class UserUseCase:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository

    async def create_user(self, username: str, email: str, password: str) -> User:
        # Implement business logic, validation, etc.
        user = User(username=username, email=email, hashed_password=self.hash_password(password))
        return await self.user_repository.create(user)

    def hash_password(self, password: str) -> str:
        # Implement password hashing
        pass

Repository Interface:

from abc import ABC, abstractmethod
from app.domain.models.user import User

class UserRepository(ABC):
    @abstractmethod
    async def create(self, user: User) -> User:
        pass

    @abstractmethod
    async def get_by_id(self, user_id: int) -> User:
        pass

    # Add other methods as needed

Repository Implementation:

from sqlalchemy.orm import Session
from app.interfaces.repositories.user_repository import UserRepository
from app.domain.models.user import User

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, db: Session):
        self.db = db

    async def create(self, user: User) -> User:
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

    async def get_by_id(self, user_id: int) -> User:
        return await self.db.query(User).filter(User.id == user_id).first()

API Endpoint:

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.api.dependencies import get_db
from app.usecases.user import UserUseCase
from app.infrastructure.repositories.sqlalchemy_user_repository import SQLAlchemyUserRepository
from app.domain.schemas.user import UserCreate, UserResponse

router = APIRouter()

@router.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    user_repository = SQLAlchemyUserRepository(db)
    user_usecase = UserUseCase(user_repository)
    created_user = await user_usecase.create_user(user.username, user.email, user.password)
    return UserResponse.from_orm(created_user)

This structure adheres to clean architecture principles by:

Separating concerns into distinct layers
Ensuring that inner layers (domain and use cases) don’t depend on outer layers
Using dependency inversion for database access (through repositories)
Keeping the domain models independent of the persistence mechanism

To further improve this structure:

Implement dependency injection for better testability
Use environment variables for configuration (stored in .env)
Implement proper error handling and custom exceptions
Add logging throughout the application
Implement unit and integration tests
Use Alembic for database migrations

This approach allows for flexibility in changing databases, easy testing of business logic, and a clear separation of concerns. It’s scalable and maintainable, making it suitable for large, complex applications.