sdcasas.dev
#fastapi #python #clean-architecture #backend #design-patterns

Clean Architecture with FastAPI: a pragmatic approach

How to structure a FastAPI project using Clean Architecture principles without over-engineering it.

5 min read

Why Clean Architecture?

Most FastAPI tutorials show everything in a single file: routes, database calls, business logic. That works for a demo. It doesn’t work when the codebase grows or a second developer joins.

Clean Architecture is not about complexity — it’s about boundaries. Where does your business logic live? Is it testable without spinning up a database? Can you swap PostgreSQL for SQLite in tests?

If the answer to any of those is “hard”, this post is for you.

The layers

┌──────────────────────────────────────────┐
│             API (FastAPI routes)          │  ← Delivery mechanism
├──────────────────────────────────────────┤
│           Application (Use Cases)         │  ← Orchestrates business logic
├──────────────────────────────────────────┤
│             Domain (Entities)             │  ← Pure Python, no dependencies
├──────────────────────────────────────────┤
│         Infrastructure (DB, Services)     │  ← Implements interfaces
└──────────────────────────────────────────┘

Rule: dependencies point inward only. Domain knows nothing about the database. Application knows nothing about HTTP. Infrastructure implements interfaces defined in Domain.

Project structure

src/
├── api/
│   ├── v1/
│   │   └── users.py          # FastAPI routers
│   └── deps.py               # Dependency injection
│
├── application/
│   └── users/
│       ├── commands.py       # Create, update, delete use cases
│       └── queries.py        # Read use cases
│
├── domain/
│   └── users/
│       ├── entity.py         # User domain model
│       ├── repository.py     # Abstract interface
│       └── exceptions.py     # Domain exceptions
│
└── infrastructure/
    └── persistence/
        ├── sqlalchemy/
        │   └── user_repo.py  # SQLAlchemy implementation
        └── models.py         # ORM models

Domain layer: pure Python

# domain/users/entity.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    id: str
    email: str
    name: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self) -> None:
        """Business rule: deactivated users cannot log in."""
        self.is_active = False
# domain/users/repository.py
from abc import ABC, abstractmethod
from .entity import User

class UserRepository(ABC):
    @abstractmethod
    async def find_by_id(self, user_id: str) -> User | None: ...

    @abstractmethod
    async def find_by_email(self, email: str) -> User | None: ...

    @abstractmethod
    async def save(self, user: User) -> User: ...

    @abstractmethod
    async def delete(self, user_id: str) -> None: ...

Application layer: use cases

# application/users/commands.py
from dataclasses import dataclass
from domain.users.entity import User
from domain.users.repository import UserRepository
from domain.users.exceptions import UserAlreadyExistsError
import uuid
from datetime import datetime

@dataclass
class CreateUserCommand:
    email: str
    name: str

class CreateUserUseCase:
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def execute(self, cmd: CreateUserCommand) -> User:
        existing = await self._repo.find_by_email(cmd.email)
        if existing:
            raise UserAlreadyExistsError(cmd.email)

        user = User(
            id=str(uuid.uuid4()),
            email=cmd.email,
            name=cmd.name,
            created_at=datetime.utcnow(),
        )
        return await self._repo.save(user)

Infrastructure layer: SQLAlchemy

# infrastructure/persistence/sqlalchemy/user_repo.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.users.entity import User
from domain.users.repository import UserRepository
from ..models import UserORM

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def find_by_id(self, user_id: str) -> User | None:
        result = await self._session.get(UserORM, user_id)
        return result.to_domain() if result else None

    async def find_by_email(self, email: str) -> User | None:
        stmt = select(UserORM).where(UserORM.email == email)
        result = await self._session.execute(stmt)
        orm = result.scalar_one_or_none()
        return orm.to_domain() if orm else None

    async def save(self, user: User) -> User:
        orm = UserORM.from_domain(user)
        self._session.add(orm)
        await self._session.flush()
        return user

    async def delete(self, user_id: str) -> None:
        orm = await self._session.get(UserORM, user_id)
        if orm:
            await self._session.delete(orm)

API layer: thin and clean

# api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from application.users.commands import CreateUserCommand, CreateUserUseCase
from domain.users.exceptions import UserAlreadyExistsError
from api.deps import get_create_user_use_case

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

class CreateUserRequest(BaseModel):
    email: str
    name: str

class UserResponse(BaseModel):
    id: str
    email: str
    name: str

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    body: CreateUserRequest,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
):
    try:
        user = await use_case.execute(CreateUserCommand(email=body.email, name=body.name))
        return UserResponse(id=user.id, email=user.email, name=user.name)
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=409, detail=str(e))

Testing without a database

This is where the payoff becomes obvious. The domain and application layers are plain Python — you can test them with a simple in-memory repository:

# tests/application/test_create_user.py
import pytest
from application.users.commands import CreateUserCommand, CreateUserUseCase
from domain.users.repository import UserRepository
from domain.users.entity import User
from domain.users.exceptions import UserAlreadyExistsError

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._users: dict[str, User] = {}

    async def find_by_id(self, user_id: str) -> User | None:
        return self._users.get(user_id)

    async def find_by_email(self, email: str) -> User | None:
        return next((u for u in self._users.values() if u.email == email), None)

    async def save(self, user: User) -> User:
        self._users[user.id] = user
        return user

    async def delete(self, user_id: str) -> None:
        self._users.pop(user_id, None)

@pytest.mark.asyncio
async def test_create_user_success():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(repo)
    user = await use_case.execute(CreateUserCommand(email="a@b.com", name="Alice"))
    assert user.email == "a@b.com"
    assert user.is_active is True

@pytest.mark.asyncio
async def test_create_user_duplicate_raises():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(repo)
    cmd = CreateUserCommand(email="a@b.com", name="Alice")
    await use_case.execute(cmd)
    with pytest.raises(UserAlreadyExistsError):
        await use_case.execute(cmd)

Fast, minimal, no database, no HTTP server. That’s the point.

Pragmatic notes

  • Don’t apply this to every endpoint. CRUD with no business logic doesn’t need all these layers.
  • Start with a simple structure and extract layers when the logic grows.
  • The test quality is the real metric, not the folder structure.

Clean Architecture is a tool. Use it where it creates clarity, not everywhere.