PROJECT_STRUCTURE.md

Detailed Project Structure Guide

Complete Directory Tree

FastAPIBluePrint/
│
├── src/                                 # Application source code
│   ├── domain/                          # Domain layer (business logic)
│   │   ├── __init__.py
│   │   ├── entities/                    # Core business objects
│   │   │   ├── __init__.py
│   │   │   ├── base_entity.py          # Base class for entities
│   │   │   └── example_entity.py       # Example: User, Product, Order
│   │   │
│   │   ├── value_objects/              # Immutable value objects
│   │   │   ├── __init__.py
│   │   │   └── example_value_object.py # Example: Email, Money, Address
│   │   │
│   │   ├── exceptions/                 # Domain-specific exceptions
│   │   │   ├── __init__.py
│   │   │   └── domain_exceptions.py    # Business rule violations
│   │   │
│   │   └── repositories/               # Repository interfaces (contracts)
│   │       ├── __init__.py
│   │       ├── base_repository.py      # Abstract base for repositories
│   │       └── example_repository.py   # Example: IUserRepository
│   │
│   ├── application/                     # Application layer (use cases)
│   │   ├── __init__.py
│   │   ├── dto/                         # Data Transfer Objects
│   │   │   ├── __init__.py
│   │   │   ├── base_dto.py             # Base DTO class
│   │   │   └── example_dto.py          # Example: UserInputDTO, UserOutputDTO
│   │   │
│   │   ├── mappers/                     # DTO to Entity mappers
│   │   │   ├── __init__.py
│   │   │   ├── base_mapper.py          # Base mapper class
│   │   │   └── example_mapper.py       # Example: UserMapper
│   │   │
│   │   └── use_cases/                   # Use case implementations
│   │       ├── __init__.py
│   │       ├── base_use_case.py        # Base use case class
│   │       └── example_use_case.py     # Example: CreateUserUseCase
│   │
│   ├── infrastructure/                  # Infrastructure layer
│   │   ├── __init__.py
│   │   ├── database/                    # Database configuration
│   │   │   ├── __init__.py
│   │   │   ├── base.py                 # Base ORM model
│   │   │   ├── session.py              # Database session management
│   │   │   └── models.py               # SQLAlchemy models
│   │   │
│   │   ├── repositories/                # Repository implementations
│   │   │   ├── __init__.py
│   │   │   └── sqlalchemy_example_repo.py  # Example: SQLAlchemyUserRepository
│   │   │
│   │   └── external_services/          # External service adapters
│   │       ├── __init__.py
│   │       └── example_adapter.py      # Example: EmailServiceAdapter
│   │
│   ├── presentation/                    # Presentation layer (FastAPI)
│   │   ├── __init__.py
│   │   ├── api/                         # API route grouping
│   │   │   ├── __init__.py
│   │   │   └── example_router.py       # Example: users router
│   │   │
│   │   ├── schemas/                     # Pydantic request/response models
│   │   │   ├── __init__.py
│   │   │   ├── base_schema.py          # Base schema configuration
│   │   │   └── example_schemas.py      # Example: CreateUserRequest, UserResponse
│   │   │
│   │   ├── dependencies.py              # Dependency injection setup
│   │   │
│   │   └── middleware/                  # Custom middleware
│   │       ├── __init__.py
│   │       ├── error_handler.py        # Global error handling
│   │       └── logging_middleware.py   # Request/response logging
│   │
│   ├── config/                          # Configuration management
│   │   ├── __init__.py
│   │   └── settings.py                  # Pydantic settings model
│   │
│   └── main.py                          # Application entry point
│
├── tests/                               # Test suite
│   ├── __init__.py
│   ├── conftest.py                      # Pytest fixtures and setup
│   │
│   ├── unit/                            # Unit tests (domain & application)
│   │   ├── __init__.py
│   │   ├── domain/
│   │   │   ├── __init__.py
│   │   │   ├── test_entities.py         # Entity logic tests
│   │   │   └── test_value_objects.py    # Value object tests
│   │   │
│   │   └── application/
│   │       ├── __init__.py
│   │       └── test_use_cases.py        # Use case tests
│   │
│   └── integration/                     # Integration tests (infrastructure & API)
│       ├── __init__.py
│       ├── repositories/
│       │   ├── __init__.py
│       │   └── test_repositories.py     # Repository implementation tests
│       │
│       └── api/
│           ├── __init__.py
│           └── test_api_endpoints.py    # API endpoint tests
│
├── docs/                                # Documentation
│   ├── ARCHITECTURE.md                  # Clean Architecture principles
│   ├── PROJECT_STRUCTURE.md             # This file
│   ├── DEVELOPMENT_GUIDE.md             # How to extend the blueprint
│   └── TESTING_STRATEGY.md              # Testing approach and examples
│
├── .env.example                         # Environment variables template
├── .gitignore                           # Git ignore rules
├── pytest.ini                           # Pytest configuration
├── pyproject.toml                       # Project metadata and dependencies
├── requirements.txt                     # Python package dependencies
└── README.md                            # Project overview

Directory Purposes

src/domain/ - Domain Layer

The innermost layer containing pure business logic with zero external dependencies.

Subdirectories:

entities/

• Purpose: Define core business objects representing domain concepts
• Files:
  • base_entity.py: Abstract base class with common entity behavior
  • example_entity.py: Concrete implementations (User, Product, Order)
• Key Rule: Entities should encapsulate business rules and invariants
• Example:

  class User(BaseEntity):
      """User entity encapsulating user-related business rules."""
      
      def __init__(self, id: int | None, email: str, name: str):
          self.id = id
          self.email = email
          self.name = name
          self._validate()
      
      def _validate(self) -> None:
          """Enforce business invariants."""
          if not self.email or "@" not in self.email:
              raise InvalidEmailError()

value_objects/

• Purpose: Represent immutable values with encapsulated logic
• Files:
  • example_value_object.py: Email, Money, Address, Status
• Key Rule: Equality based on values, not identity
• Use When: A concept is immutable and validated frequently (Email, PhoneNumber)

exceptions/

• Purpose: Domain-specific exceptions for business rule violations
• Files:
  • domain_exceptions.py: UserAlreadyExistsError, InsufficientFundsError
• Key Rule: These are not HTTP errors; they're domain violations
• Handled By: Application layer, converted to HTTP errors by presentation layer

repositories/

• Purpose: Define contracts for data access without implementation
• Files:
  • base_repository.py: Base interface for all repositories
  • example_repository.py: IUserRepository, IProductRepository
• Key Rule: Depend on these interfaces in use cases, not concrete implementations
• Example:

  class IUserRepository(ABC):
      """Contract for user persistence - no implementation."""
      
      @abstractmethod
      async def find_by_id(self, user_id: int) -> User | None:
          """Find user by ID. Returns None if not found."""
          pass
      
      @abstractmethod
      async def save(self, user: User) -> User:
          """Persist user and return with ID populated."""
          pass

---

src/application/ - Application Layer

Orchestrates domain logic and implements use cases. Contains no database or framework code.

Subdirectories:

dto/

• Purpose: Define data structures for transferring data between layers
• Files:
  • base_dto.py: Base DTO configuration
  • example_dto.py: CreateUserInputDTO, UserOutputDTO
• Key Rule: DTOs have no business logic, just data structure
• Separation:
  • InputDTO: Data coming from presentation layer
  • OutputDTO: Data returned to presentation layer
• Example:

  class CreateUserInputDTO:
      """Data required to create a user."""
      email: str
      name: str
  
  class UserOutputDTO:
      """User data returned to presentation layer."""
      id: int
      email: str
      name: str
      created_at: datetime

mappers/

• Purpose: Convert between domain entities and DTOs
• Files:
  • base_mapper.py: Base mapper abstract class
  • example_mapper.py: UserMapper (entity ↔ DTO)
• Key Rule: Mappers ensure DTOs don't leak domain details
• Example:

  class UserMapper:
      @staticmethod
      def to_domain(dto: CreateUserInputDTO) -> User:
          return User(id=None, email=dto.email, name=dto.name)
      
      @staticmethod
      def to_output_dto(user: User) -> UserOutputDTO:
          return UserOutputDTO(
              id=user.id,
              email=user.email,
              name=user.name,
              created_at=user.created_at
          )

use_cases/

• Purpose: Implement application-specific business rules and orchestration
• Files:
  • base_use_case.py: Base class with common patterns
  • example_use_case.py: CreateUserUseCase, UpdateUserUseCase
• Key Rule: Use cases depend on repository interfaces, not implementations
• Responsibilities:
  • Orchestrate domain objects
  • Check business rules
  • Coordinate repositories
  • Handle transactions
• Example:

  class CreateUserUseCase:
      """Use case for creating a new user."""
      
      def __init__(self, user_repository: IUserRepository):
          self.user_repository = user_repository
      
      async def execute(self, input_dto: CreateUserInputDTO) -> UserOutputDTO:
          # Check business rule
          existing = await self.user_repository.find_by_email(input_dto.email)
          if existing:
              raise UserAlreadyExistsError(input_dto.email)
          
          # Create entity
          user = UserMapper.to_domain(input_dto)
          
          # Persist through repository interface
          created_user = await self.user_repository.save(user)
          
          # Return DTO
          return UserMapper.to_output_dto(created_user)

---

src/infrastructure/ - Infrastructure Layer

Implements technical details: databases, external services, ORM models.

Subdirectories:

database/

• Purpose: Database setup, ORM models, session management
• Files:
  • base.py: SQLAlchemy declarative base
  • session.py: Database connection and session factory
  • models.py: ORM model definitions (separate from domain entities)
• Key Rule: ORM models are infrastructure details; don't use directly in domain/application
• Example:

  # models.py - SQLAlchemy model (infrastructure detail)
  class UserModel(Base):
      """Database representation of a user."""
      __tablename__ = "users"
      
      id = Column(Integer, primary_key=True)
      email = Column(String, unique=True, index=True)
      name = Column(String)
      created_at = Column(DateTime, default=datetime.utcnow)
  
  # session.py - Session factory
  async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
      async with AsyncSessionLocal() as session:
          yield session

repositories/

• Purpose: Concrete implementations of repository interfaces
• Files:
  • sqlalchemy_example_repo.py: SQLAlchemyUserRepository
• Key Rule: Implement domain repository interfaces; translate between ORM models and domain entities
• Switching Databases: To change from PostgreSQL to MongoDB, implement new repository without touching domain/application
• Example:

  class SQLAlchemyUserRepository(IUserRepository):
      """Concrete user repository using SQLAlchemy."""
      
      def __init__(self, session: AsyncSession):
          self.session = session
      
      async def find_by_id(self, user_id: int) -> User | None:
          result = await self.session.execute(
              select(UserModel).where(UserModel.id == user_id)
          )
          model = result.scalar_one_or_none()
          return self._to_domain(model) if model else None
      
      async def save(self, user: User) -> User:
          model = self._to_model(user)
          self.session.add(model)
          await self.session.commit()
          return self._to_domain(model)
      
      @staticmethod
      def _to_domain(model: UserModel) -> User:
          return User(id=model.id, email=model.email, name=model.name)
      
      @staticmethod
      def _to_model(user: User) -> UserModel:
          return UserModel(id=user.id, email=user.email, name=user.name)

external_services/

• Purpose: Adapters for external APIs and services
• Files:
  • example_adapter.py: EmailServiceAdapter, PaymentGatewayAdapter
• Key Rule: Translate external APIs to domain concepts
• Isolation: If external service changes, only this layer needs updates
• Example:

  class EmailServiceAdapter:
      """Adapter for email service (SendGrid, Mailgun, etc)."""
      
      def __init__(self, api_key: str):
          self.api_key = api_key
      
      async def send_welcome_email(self, email: str, name: str) -> bool:
          """Send welcome email using external service."""
          # External service API calls
          return True

---

src/presentation/ - Presentation Layer

FastAPI-specific code: routers, request/response schemas, middleware.

Subdirectories:

api/

• Purpose: FastAPI routers defining HTTP endpoints
• Files:
  • example_router.py: Users router, Products router
• Key Rule: Routers are thin; they delegate to use cases
• Responsibilities:
  • Define HTTP paths and methods
  • Handle dependencies injection
  • Validate requests with schemas
  • Catch exceptions and return appropriate HTTP responses
• Example:

  router = APIRouter(prefix="/api/users", tags=["users"])
  
  def get_create_user_use_case(
      session: AsyncSession = Depends(get_db_session)
  ) -> CreateUserUseCase:
      repository = SQLAlchemyUserRepository(session)
      return CreateUserUseCase(repository)
  
  @router.post("/", status_code=201, response_model=UserResponse)
  async def create_user(
      request: CreateUserRequest,
      use_case: CreateUserUseCase = Depends(get_create_user_use_case)
  ) -> UserResponse:
      try:
          result = await use_case.execute(
              CreateUserInputDTO(email=request.email, name=request.name)
          )
          return UserResponse.from_dto(result)
      except UserAlreadyExistsError as e:
          raise HTTPException(status_code=409, detail=str(e))

schemas/

• Purpose: Pydantic models for request/response validation and documentation
• Files:
  • base_schema.py: Base schema configuration
  • example_schemas.py: CreateUserRequest, UserResponse
• Key Rule: Schemas validate input and serialize output; they're API contracts
• Separation: Input schemas (CreateUserRequest) vs Output schemas (UserResponse)
• Example:

  class CreateUserRequest(BaseModel):
      """Request schema for creating a user."""
      email: EmailStr
      name: str = Field(..., min_length=1, max_length=100)
  
  class UserResponse(BaseModel):
      """Response schema for user data."""
      id: int
      email: str
      name: str
      created_at: datetime
      
      @classmethod
      def from_dto(cls, dto: UserOutputDTO) -> "UserResponse":
          return cls(
              id=dto.id,
              email=dto.email,
              name=dto.name,
              created_at=dto.created_at
          )

dependencies.py

• Purpose: Dependency injection configuration
• Content:
  • Database session providers
  • Repository providers
  • Use case providers
  • Authentication/authorization
• Key Rule: All dependencies created here; routers depend on these functions
• Example:

  async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
      async with AsyncSessionLocal() as session:
          yield session
  
  def get_user_repository(
      session: AsyncSession = Depends(get_db_session)
  ) -> IUserRepository:
      return SQLAlchemyUserRepository(session)
  
  def get_create_user_use_case(
      repository: IUserRepository = Depends(get_user_repository)
  ) -> CreateUserUseCase:
      return CreateUserUseCase(repository)

middleware/

• Purpose: Cross-cutting concerns
• Files:
  • error_handler.py: Global exception handling
  • logging_middleware.py: Request/response logging
• Example:

  # error_handler.py
  @app.exception_handler(UserAlreadyExistsError)
  async def user_exists_exception_handler(request: Request, exc: UserAlreadyExistsError):
      return JSONResponse(
          status_code=409,
          content={"detail": str(exc)}
      )

---

src/config/ - Configuration Management

Files:

• settings.py: Pydantic-based settings model

Purpose: Centralize environment-specific configuration

Example:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """Application configuration loaded from environment."""
    
    # Database
    database_url: str = "sqlite:///./test.db"
    
    # API
    api_title: str = "FastAPI Blueprint"
    api_version: str = "1.0.0"
    
    # Logging
    log_level: str = "INFO"
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()

---

src/main.py - Application Entry Point

Purpose: Create and configure FastAPI application

Responsibilities:

• Initialize FastAPI app
• Register routers
• Setup middleware
• Setup exception handlers
• Initialize logging

Example:

from fastapi import FastAPI
from src.presentation.api import example_router
from src.config.settings import settings

app = FastAPI(
    title=settings.api_title,
    version=settings.api_version
)

app.include_router(example_router.router)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("src.main:app", reload=True)

---

tests/ - Test Suite

conftest.py

• Purpose: Pytest configuration and shared fixtures
• Content:
  • Database fixtures for testing
  • Mock repositories
  • Test client setup
• Example:

  @pytest.fixture
  async def mock_user_repository() -> IUserRepository:
      """Mock repository for use case tests."""
      class MockUserRepository(IUserRepository):
          async def find_by_email(self, email: str) -> User | None:
              return None
          async def save(self, user: User) -> User:
              return User(id=1, email=user.email, name=user.name)
      
      return MockUserRepository()

tests/unit/

• Purpose: Test business logic in isolation (domain and application layers)
• No external dependencies: No database, no HTTP calls
• Files:
  • domain/test_entities.py: Entity validation, invariants
  • application/test_use_cases.py: Use case logic with mock repositories
• Strategy: Test the "what", not the "how"
• Example:

  @pytest.mark.asyncio
  async def test_create_user_success(mock_user_repository):
      use_case = CreateUserUseCase(mock_user_repository)
      
      result = await use_case.execute(
          CreateUserInputDTO(email="test@example.com", name="Test")
      )
      
      assert result.email == "test@example.com"
      assert result.id == 1

tests/integration/

• Purpose: Test infrastructure and API integration
• With dependencies: Real database (in-memory SQLite), real HTTP calls
• Files:
  • repositories/test_repositories.py: Repository implementations
  • api/test_api_endpoints.py: Full HTTP endpoint tests
• Example:

  @pytest.mark.asyncio
  async def test_create_user_endpoint(client, db_session):
      response = client.post(
          "/api/users/",
          json={"email": "test@example.com", "name": "Test"}
      )
      
      assert response.status_code == 201
      assert response.json()["email"] == "test@example.com"

---

Configuration Files

.env.example

Template for environment variables. Copy to .env for local development.

DATABASE_URL=sqlite+aiosqlite:///./test.db
API_TITLE=FastAPI Blueprint
LOG_LEVEL=INFO

pyproject.toml

Project metadata and dependencies.

[project]
name = "fastapi-blueprint"
version = "1.0.0"
description = "Clean Architecture blueprint for FastAPI"
requires-python = ">=3.10"

[project.optional-dependencies]
dev = ["pytest", "pytest-asyncio", "black", "ruff"]

pytest.ini

Pytest configuration.

[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py

requirements.txt

Python dependencies.

fastapi==0.104.1
uvicorn==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.0
pydantic-settings==2.1.0
python-dotenv==1.0.0
pytest==7.4.3
pytest-asyncio==0.21.1

---

Adding a New Feature

Follow these steps to add a new domain feature (e.g., creating a Products feature):

1. Define Domain Layer

Create src/domain/entities/product.py:

class Product(BaseEntity):
    def __init__(self, id: int | None, name: str, price: float):
        self.id = id
        self.name = name
        self.price = price
        self._validate()
    
    def _validate(self) -> None:
        if self.price < 0:
            raise InvalidPriceError()

Create src/domain/repositories/product_repository.py:

class IProductRepository(ABC):
    @abstractmethod
    async def find_by_id(self, product_id: int) -> Product | None: pass
    
    @abstractmethod
    async def save(self, product: Product) -> Product: pass

2. Define Application Layer

Create src/application/use_cases/create_product_use_case.py:

class CreateProductUseCase:
    def __init__(self, product_repository: IProductRepository):
        self.product_repository = product_repository
    
    async def execute(self, name: str, price: float) -> ProductOutputDTO:
        product = Product(id=None, name=name, price=price)
        created = await self.product_repository.save(product)
        return ProductMapper.to_output_dto(created)

3. Implement Infrastructure

Create src/infrastructure/database/models.py (add):

class ProductModel(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Float)

Create src/infrastructure/repositories/sqlalchemy_product_repo.py:

class SQLAlchemyProductRepository(IProductRepository):
    # Implementation similar to user repository

4. Create Presentation Layer

Create src/presentation/api/product_router.py:

@router.post("/", status_code=201)
async def create_product(
    request: CreateProductRequest,
    use_case: CreateProductUseCase = Depends(get_create_product_use_case)
):
    result = await use_case.execute(request.name, request.price)
    return ProductResponse.from_dto(result)

Create src/presentation/schemas/product_schemas.py:

class CreateProductRequest(BaseModel):
    name: str
    price: float

class ProductResponse(BaseModel):
    id: int
    name: str
    price: float

5. Write Tests

Create tests/unit/application/test_product_use_cases.py Create tests/integration/api/test_product_endpoints.py

---

Key Takeaway: Each layer has a clear responsibility. Add features by following this pattern without modifying existing code.