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:

1. Domain Layer:

2. Use Case Layer:

3. Interface Adapters Layer:

4. Infrastructure Layer:

5. Main Application: Here’s how to implement some key components:

6. 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)

7. 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

8. 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

9. 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()

10. 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:

1. Separating concerns into distinct layers

2. Ensuring that inner layers (domain and use cases) don’t depend on outer layers

3. Using dependency inversion for database access (through repositories)

4. Keeping the domain models independent of the persistence mechanism To further improve this structure:

5. Implement dependency injection for better testability

6. Use environment variables for configuration (stored in .env)

7. Implement proper error handling and custom exceptions

8. Add logging throughout the application

9. Implement unit and integration tests

10. 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.