Fastapi Clean Architecture

FastAPI Clean Architecture Skill

Overview

This skill guides you in building FastAPI applications following Clean Architecture principles, based on production patterns from enterprise financial systems. It emphasizes proper layer separation, dependency injection, repository patterns, and comprehensive testing strategies.

When to Use This Skill

• Starting a new FastAPI project requiring clean architecture
• Refactoring existing FastAPI code for better maintainability
• Implementing domain-driven design with FastAPI
• Building testable, scalable backend services
• Integrating multiple data sources (PostgreSQL, Redis, external APIs)

Core Architecture Principles

Three-Layer Architecture

src/
├── api/              # API Layer (Controllers, Routes, DTOs)
├── domain/           # Domain Layer (Business Logic, Entities, Services)
└── infra/            # Infrastructure Layer (Database, External Services)

Layer Responsibilities:

1. API Layer (src/api/)

• HTTP endpoints and routing
• Request/Response DTOs (Pydantic models)
• Input validation
• Authentication/Authorization middleware
• Error handling and HTTP status codes

1. Domain Layer (src/domain/)

• Business entities and value objects
• Service interfaces (abstract classes)
• Business rules and validation
• Domain exceptions
• Pure business logic (framework-agnostic)

1. Infrastructure Layer (src/infra/)

• Database repositories (SQLAlchemy)
• External API adapters
• Cache implementations (Redis)
• File storage adapters
• Concrete service implementations

Dependency Flow

API Layer → Domain Layer ← Infrastructure Layer

Critical Rules:

• API depends on Domain (imports domain services/entities)
• Infrastructure implements Domain interfaces
• Domain NEVER imports from API or Infrastructure
• Use dependency injection to wire everything together

Project Structure Template

project-name/
├── src/
│   ├── api/
│   │   ├── path/              # Route modules
│   │   │   ├── users.py
│   │   │   ├── auth.py
│   │   │   └── financial.py
│   │   ├── middlewares/       # Custom middleware
│   │   │   ├── auth.py
│   │   │   └── error_handler.py
│   │   └── schemas/           # Request/Response DTOs
│   │       ├── user.py
│   │       └── financial.py
│   ├── domain/
│   │   ├── modules/           # Business modules
│   │   │   ├── user/
│   │   │   │   ├── entity.py       # User entity
│   │   │   │   ├── service.py      # IUserService (abstract)
│   │   │   │   ├── repository.py   # IUserRepository (abstract)
│   │   │   │   └── exceptions.py   # Domain exceptions
│   │   │   └── financial/
│   │   │       ├── entity.py
│   │   │       ├── service.py
│   │   │       └── value_objects.py
│   │   └── shared/            # Shared domain logic
│   │       ├── base_entity.py
│   │       └── exceptions.py
│   ├── infra/
│   │   ├── database/
│   │   │   ├── alchemist/     # SQLAlchemy implementations
│   │   │   │   ├── modules/
│   │   │   │   │   ├── user/
│   │   │   │   │   │   └── repository.py  # UserRepository
│   │   │   │   │   └── financial/
│   │   │   │   │       └── repository.py
│   │   │   │   └── session.py
│   │   │   └── migrations/    # Database migrations
│   │   ├── adapters/          # External service adapters
│   │   │   ├── auth/
│   │   │   │   └── sso_adapter.py
│   │   │   └── payment/
│   │   │       └── payment_gateway.py
│   │   ├── cache/             # Redis implementations
│   │   │   └── redis_cache.py
│   │   └── services/          # Service implementations
│   │       ├── user_service.py
│   │       └── financial_service.py
│   ├── config/
│   │   ├── settings.py        # Environment configuration
│   │   ├── dependency.py      # DI Container
│   │   └── database.py        # DB configuration
│   └── main.py                # FastAPI app entry point
├── tests/
│   ├── api/                   # API integration tests
│   ├── domain/                # Domain unit tests
│   ├── infra/                 # Infrastructure tests
│   └── conftest.py            # Pytest fixtures
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
└── README.md

Implementation Patterns

1. Domain Entity Pattern

# src/domain/modules/user/entity.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Domain entity representing a user.

    Pure business logic with no framework dependencies.
    """
    id: int | None
    cpf: str
    name: str
    email: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self) -> None:
        """Business rule: Deactivate user account."""
        if not self.is_active:
            raise UserAlreadyInactiveError(f"User {self.cpf} is already inactive")
        self.is_active = False

    def validate_cpf(self) -> bool:
        """Business rule: Validate CPF format."""
        cleaned = ''.join(filter(str.isdigit, self.cpf))
        return len(cleaned) == 11 and not all(c == cleaned[0] for c in cleaned)

2. Repository Interface Pattern

# src/domain/modules/user/repository.py
from abc import ABC, abstractmethod
from typing import List

from src.domain.modules.user.entity import User

class IUserRepository(ABC):
    """Abstract repository interface in domain layer.

    Defines contract without implementation details.
    """

    @abstractmethod
    async def get_by_id(self, user_id: int) -> User | None:
        """Retrieve user by ID."""
        pass

    @abstractmethod
    async def get_by_cpf(self, cpf: str) -> User | None:
        """Retrieve user by CPF."""
        pass

    @abstractmethod
    async def create(self, user: User) -> User:
        """Create new user."""
        pass

    @abstractmethod
    async def update(self, user: User) -> User:
        """Update existing user."""
        pass

    @abstractmethod
    async def list_active(self, limit: int = 100) -> List[User]:
        """List all active users."""
        pass

3. Service Interface Pattern

# src/domain/modules/user/service.py
from abc import ABC, abstractmethod
from typing import List

from src.domain.modules.user.entity import User

class IUserService(ABC):
    """Abstract service interface defining business operations."""

    @abstractmethod
    async def register_user(self, cpf: str, name: str, email: str) -> User:
        """Register new user with validation."""
        pass

    @abstractmethod
    async def deactivate_user(self, cpf: str) -> User:
        """Deactivate user account."""
        pass

    @abstractmethod
    async def get_user_profile(self, cpf: str) -> User:
        """Get user profile by CPF."""
        pass

4. Concrete Repository Implementation

# src/infra/database/alchemist/modules/user/repository.py
from typing import List
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.modules.user.entity import User
from src.domain.modules.user.repository import IUserRepository
from src.infra.database.alchemist.models import UserModel

class UserRepository(IUserRepository):
    """SQLAlchemy implementation of user repository.

    Handles database operations and entity mapping.
    """

    def __init__(self, session: AsyncSession):
        self._session = session

    async def get_by_id(self, user_id: int) -> User | None:
        """Retrieve user by ID."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user_id)
        )
        model = result.scalar_one_or_none()
        return self._to_entity(model) if model else None

    async def get_by_cpf(self, cpf: str) -> User | None:
        """Retrieve user by CPF."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.cpf == cpf)
        )
        model = result.scalar_one_or_none()
        return self._to_entity(model) if model else None

    async def create(self, user: User) -> User:
        """Create new user."""
        model = UserModel(
            cpf=user.cpf,
            name=user.name,
            email=user.email,
            is_active=user.is_active,
        )
        self._session.add(model)
        await self._session.flush()
        await self._session.refresh(model)
        return self._to_entity(model)

    async def update(self, user: User) -> User:
        """Update existing user."""
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user.id)
        )
        model = result.scalar_one()
        model.name = user.name
        model.email = user.email
        model.is_active = user.is_active
        await self._session.flush()
        await self._session.refresh(model)
        return self._to_entity(model)

    async def list_active(self, limit: int = 100) -> List[User]:
        """List all active users."""
        result = await self._session.execute(
            select(UserModel)
            .where(UserModel.is_active == True)  # noqa: E712
            .limit(limit)
        )
        models = result.scalars().all()
        return [self._to_entity(model) for model in models]

    def _to_entity(self, model: UserModel) -> User:
        """Convert database model to domain entity."""
        return User(
            id=model.id,
            cpf=model.cpf,
            name=model.name,
            email=model.email,
            created_at=model.created_at,
            is_active=model.is_active,
        )

5. Concrete Service Implementation

# src/infra/services/user_service.py
from src.domain.modules.user.entity import User
from src.domain.modules.user.service import IUserService
from src.domain.modules.user.repository import IUserRepository
from src.domain.modules.user.exceptions import (
    UserAlreadyExistsError,
    UserNotFoundError,
    InvalidCPFError,
)

class UserService(IUserService):
    """Concrete implementation of user service.

    Orchestrates business logic using repository.
    """

    def __init__(self, user_repository: IUserRepository):
        self._repository = user_repository

    async def register_user(self, cpf: str, name: str, email: str) -> User:
        """Register new user with validation."""
        # Check if user already exists
        existing_user = await self._repository.get_by_cpf(cpf)
        if existing_user:
            raise UserAlreadyExistsError(f"User with CPF {cpf} already exists")

        # Create and validate entity
        user = User(
            id=None,
            cpf=cpf,
            name=name,
            email=email,
            created_at=datetime.now(),
            is_active=True,
        )

        if not user.validate_cpf():
            raise InvalidCPFError(f"Invalid CPF: {cpf}")

        # Persist
        return await self._repository.create(user)

    async def deactivate_user(self, cpf: str) -> User:
        """Deactivate user account."""
        user = await self._repository.get_by_cpf(cpf)
        if not user:
            raise UserNotFoundError(f"User with CPF {cpf} not found")

        user.deactivate()  # Business logic in entity
        return await self._repository.update(user)

    async def get_user_profile(self, cpf: str) -> User:
        """Get user profile by CPF."""
        user = await self._repository.get_by_cpf(cpf)
        if not user:
            raise UserNotFoundError(f"User with CPF {cpf} not found")
        return user

6. Dependency Injection Container

# src/config/dependency.py
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject

from src.infra.database.alchemist.session import get_session
from src.infra.database.alchemist.modules.user.repository import UserRepository
from src.infra.services.user_service import UserService

class Container(containers.DeclarativeContainer):
    """Dependency injection container.

    Wires together all dependencies.
    """

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.path.users",
            "src.api.path.auth",
        ]
    )

    # Database
    db_session = providers.Resource(get_session)

    # Repositories
    user_repository = providers.Factory(
        UserRepository,
        session=db_session,
    )

    # Services
    user_service = providers.Factory(
        UserService,
        user_repository=user_repository,
    )

7. API Endpoint Implementation

# src/api/path/users.py
from fastapi import APIRouter, Depends, status
from dependency_injector.wiring import inject, Provide

from src.api.schemas.user import UserCreateRequest, UserResponse
from src.domain.modules.user.service import IUserService
from src.domain.modules.user.exceptions import UserAlreadyExistsError, InvalidCPFError
from src.config.dependency import Container

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

@router.post(
    "/",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
@inject
async def create_user(
    request: UserCreateRequest,
    user_service: IUserService = Depends(Provide[Container.user_service]),
):
    """Create new user endpoint.

    Delegates to service layer for business logic.
    """
    try:
        user = await user_service.register_user(
            cpf=request.cpf,
            name=request.name,
            email=request.email,
        )
        return UserResponse.from_entity(user)
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=409, detail=str(e))
    except InvalidCPFError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/{cpf}", response_model=UserResponse)
@inject
async def get_user(
    cpf: str,
    user_service: IUserService = Depends(Provide[Container.user_service]),
):
    """Get user by CPF."""
    user = await user_service.get_user_profile(cpf)
    return UserResponse.from_entity(user)

8. Pydantic DTOs (Request/Response)

# src/api/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field

from src.domain.modules.user.entity import User

class UserCreateRequest(BaseModel):
    """Request DTO for user creation."""
    cpf: str = Field(..., min_length=11, max_length=14)
    name: str = Field(..., min_length=3, max_length=100)
    email: EmailStr

class UserResponse(BaseModel):
    """Response DTO for user data."""
    id: int
    cpf: str
    name: str
    email: str
    created_at: datetime
    is_active: bool

    @classmethod
    def from_entity(cls, user: User) -> "UserResponse":
        """Convert domain entity to DTO."""
        return cls(
            id=user.id,
            cpf=user.cpf,
            name=user.name,
            email=user.email,
            created_at=user.created_at,
            is_active=user.is_active,
        )

Testing Strategy

Unit Tests (Domain Layer)

# tests/domain/modules/user/test_entity.py
import pytest
from src.domain.modules.user.entity import User
from src.domain.modules.user.exceptions import UserAlreadyInactiveError

def test_user_deactivation():
    """Test user deactivation business rule."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="test@example.com",
        created_at=datetime.now(),
        is_active=True,
    )

    user.deactivate()
    assert user.is_active is False

def test_user_already_inactive_raises_error():
    """Test deactivating already inactive user raises error."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Test User",
        email="test@example.com",
        created_at=datetime.now(),
        is_active=False,
    )

    with pytest.raises(UserAlreadyInactiveError):
        user.deactivate()

def test_cpf_validation():
    """Test CPF validation logic."""
    user = User(
        id=1,
        cpf="12345678901",
        name="Tes

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [rafaelkamimura](https://github.com/rafaelkamimura)
- **Source:** [rafaelkamimura/claude-tools](https://github.com/rafaelkamimura/claude-tools)
- **License:** MIT

Install and usage instructions live in the source repository linked above.