Production-Ready FastAPI Project Structure (2026 Guide)

Your FastAPI tutorial app worked great — until it didn't. Fifty endpoints crammed into one file, database sessions leaking across requests, and tests that require the entire application stack to run. Sound familiar? The fix isn't a framework change — it's a structure change.

This guide covers how to structure a FastAPI project that scales from MVP to production — including directory layout, dependency injection, configuration management, testing, and containerized deployment.

The Project Structure

Here's the full structure we'll build. Every directory has a purpose.

myapp/
├── app/
│   ├── __init__.py
│   ├── main.py                 # Application factory
│   ├── config.py               # Settings management
│   ├── database.py             # Database session setup
│   ├── dependencies.py         # Shared dependencies
│   ├── exceptions.py           # Custom exception handlers
│   ├── middleware.py            # Custom middleware
│   ├── api/
│   │   ├── __init__.py
│   │   ├── router.py           # Root router aggregation
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py       # v1 router
│   │   │   ├── users.py        # User endpoints
│   │   │   ├── orders.py       # Order endpoints
│   │   │   └── auth.py         # Auth endpoints
│   │   └── v2/
│   │       └── ...
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py             # SQLAlchemy base
│   │   ├── user.py
│   │   └── order.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py             # Pydantic schemas
│   │   └── order.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py     # Business logic
│   │   └── order_service.py
│   ├── repositories/
│   │   ├── __init__.py
│   │   ├── base.py             # Generic repository
│   │   ├── user_repo.py
│   │   └── order_repo.py
│   └── core/
│       ├── __init__.py
│       ├── security.py         # JWT, hashing
│       └── logging.py          # Structured logging
├── migrations/
│   └── versions/
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_orders.py
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
├── alembic.ini
└── .env.example

Why This Layout

• api/ — Route handlers only. No business logic. A thin layer that calls services.
• services/ — Business logic, orchestration, validation. Testable without HTTP.
• repositories/ — Database access. Single responsibility: CRUD operations.
• schemas/ — Pydantic models for request/response validation.
• models/ — SQLAlchemy ORM models. Separate from Pydantic schemas.
• core/ — Cross-cutting concerns: security, logging, utilities.

This separation means you can test business logic without spinning up a web server, swap databases without touching business rules, and version your API without duplicating code.

Configuration Management

Configuration is the first thing to get right. Use Pydantic Settings for type-safe, validated configuration that fails fast on missing values.

# app/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # Application
    app_name: str = "MyApp"
    app_version: str = "1.0.0"
    debug: bool = False
    environment: str = "production"

    # Database
    database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/myapp"
    database_pool_size: int = 20
    database_max_overflow: int = 10

    # Authentication
    secret_key: str
    access_token_expire_minutes: int = 30
    refresh_token_expire_days: int = 7

    # External Services
    redis_url: str = "redis://localhost:6379/0"
    smtp_host: str = "localhost"
    smtp_port: int = 587

    # CORS
    allowed_origins: list[str] = ["http://localhost:3000"]

    @property
    def async_database_url(self) -> str:
        return self.database_url.replace(
            "postgresql://", "postgresql+asyncpg://"
        )


@lru_cache
def get_settings() -> Settings:
    return Settings()

The @lru_cache decorator ensures settings are loaded once and reused — no repeated file reads or environment lookups on every request.

Application Factory

Use an application factory pattern instead of a global app instance. This makes testing dramatically simpler because each test run gets a fresh application.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import get_settings
from app.database import engine, sessionmanager
from app.api.router import api_router
from app.exceptions import register_exception_handlers
from app.middleware import RequestLoggingMiddleware


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle: startup and shutdown."""
    settings = get_settings()

    # Startup
    await sessionmanager.init(settings.database_url)
    yield

    # Shutdown
    await sessionmanager.close()


def create_app() -> FastAPI:
    settings = get_settings()

    app = FastAPI(
        title=settings.app_name,
        version=settings.app_version,
        docs_url="/docs" if settings.debug else None,
        redoc_url="/redoc" if settings.debug else None,
        lifespan=lifespan,
    )

    # Middleware (order matters — last added = first executed)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.allowed_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    app.add_middleware(RequestLoggingMiddleware)

    # Register routes
    app.include_router(api_router, prefix="/api")

    # Register exception handlers
    register_exception_handlers(app)

    return app


app = create_app()

Database Layer with Async SQLAlchemy

# app/database.py
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine
)


class DatabaseSessionManager:
    """Manages async database sessions with proper lifecycle."""

    def __init__(self):
        self._engine = None
        self._sessionmaker = None

    async def init(self, database_url: str):
        self._engine = create_async_engine(
            database_url,
            pool_size=20,
            max_overflow=10,
            pool_pre_ping=True,
            echo=False,
        )
        self._sessionmaker = async_sessionmaker(
            bind=self._engine,
            class_=AsyncSession,
            expire_on_commit=False,
        )

    async def close(self):
        if self._engine:
            await self._engine.dispose()

    @asynccontextmanager
    async def session(self):
        if self._sessionmaker is None:
            raise RuntimeError("DatabaseSessionManager is not initialized")

        async with self._sessionmaker() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise


sessionmanager = DatabaseSessionManager()


async def get_db_session():
    """FastAPI dependency for database sessions."""
    async with sessionmanager.session() as session:
        yield session

Repository Pattern

Repositories abstract database operations. Each entity gets its own repository that inherits common CRUD from a generic base.

# app/repositories/base.py
from typing import Generic, TypeVar, Type
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.base import Base

ModelType = TypeVar("ModelType", bound=Base)


class BaseRepository(Generic[ModelType]):
    """Generic repository with common CRUD operations."""

    def __init__(self, model: Type[ModelType], session: AsyncSession):
        self.model = model
        self.session = session

    async def get_by_id(self, id: int) -> ModelType | None:
        return await self.session.get(self.model, id)

    async def get_all(
        self, offset: int = 0, limit: int = 100
    ) -> list[ModelType]:
        query = select(self.model).offset(offset).limit(limit)
        result = await self.session.execute(query)
        return list(result.scalars().all())

    async def create(self, **kwargs) -> ModelType:
        instance = self.model(**kwargs)
        self.session.add(instance)
        await self.session.flush()
        await self.session.refresh(instance)
        return instance

    async def update(
        self, id: int, **kwargs
    ) -> ModelType | None:
        instance = await self.get_by_id(id)
        if not instance:
            return None
        for key, value in kwargs.items():
            setattr(instance, key, value)
        await self.session.flush()
        await self.session.refresh(instance)
        return instance

    async def delete(self, id: int) -> bool:
        instance = await self.get_by_id(id)
        if not instance:
            return False
        await self.session.delete(instance)
        await self.session.flush()
        return True

    async def count(self) -> int:
        query = select(func.count()).select_from(self.model)
        result = await self.session.execute(query)
        return result.scalar_one()

# app/repositories/user_repo.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.repositories.base import BaseRepository


class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(User, session)

    async def get_by_email(self, email: str) -> User | None:
        query = select(User).where(User.email == email)
        result = await self.session.execute(query)
        return result.scalar_one_or_none()

    async def get_active_users(
        self, offset: int = 0, limit: int = 100
    ) -> list[User]:
        query = (
            select(User)
            .where(User.is_active.is_(True))
            .offset(offset)
            .limit(limit)
            .order_by(User.created_at.desc())
        )
        result = await self.session.execute(query)
        return list(result.scalars().all())

Service Layer

Services contain business logic and orchestrate repositories. They're the layer where rules like "no duplicate emails" live — separate from both HTTP handling and database access.

# app/services/user_service.py
from fastapi import HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserUpdate, UserResponse
from app.core.security import hash_password, verify_password


class UserService:
    def __init__(self, session: AsyncSession):
        self.repo = UserRepository(session)

    async def create_user(self, data: UserCreate) -> UserResponse:
        # Business rule: check for duplicate email
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Email already registered",
            )

        user = await self.repo.create(
            email=data.email,
            name=data.name,
            hashed_password=hash_password(data.password),
        )
        return UserResponse.model_validate(user)

    async def get_user(self, user_id: int) -> UserResponse:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="User not found",
            )
        return UserResponse.model_validate(user)

    async def list_users(
        self, offset: int = 0, limit: int = 100
    ) -> list[UserResponse]:
        users = await self.repo.get_active_users(offset, limit)
        return [UserResponse.model_validate(u) for u in users]

API Routes (Thin Handlers)

Route handlers should be thin. They parse requests, call services, and return responses — nothing more.

# app/api/v1/users.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db_session
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService

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


def get_user_service(
    session: AsyncSession = Depends(get_db_session),
) -> UserService:
    return UserService(session)


@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    data: UserCreate,
    service: UserService = Depends(get_user_service),
):
    return await service.create_user(data)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    return await service.get_user(user_id)


@router.get("/", response_model=list[UserResponse])
async def list_users(
    offset: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    service: UserService = Depends(get_user_service),
):
    return await service.list_users(offset, limit)

# app/api/router.py
from fastapi import APIRouter
from app.api.v1.router import router as v1_router

api_router = APIRouter()
api_router.include_router(v1_router, prefix="/v1")

Custom Exception Handling

Consistent error responses across your entire API — no more guessing the response shape when something fails.

# app/exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError


class AppException(Exception):
    def __init__(
        self, status_code: int, detail: str, error_code: str | None = None
    ):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code


def register_exception_handlers(app: FastAPI):
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": {
                    "message": exc.detail,
                    "code": exc.error_code,
                    "status": exc.status_code,
                }
            },
        )

    @app.exception_handler(ValidationError)
    async def validation_exception_handler(
        request: Request, exc: ValidationError
    ):
        return JSONResponse(
            status_code=422,
            content={
                "error": {
                    "message": "Validation failed",
                    "code": "VALIDATION_ERROR",
                    "details": exc.errors(),
                }
            },
        )

    @app.exception_handler(Exception)
    async def general_exception_handler(request: Request, exc: Exception):
        return JSONResponse(
            status_code=500,
            content={
                "error": {
                    "message": "Internal server error",
                    "code": "INTERNAL_ERROR",
                }
            },
        )

Testing Setup

Testing is where the layered architecture pays off. You can test services without HTTP, and routes with a lightweight test client — no production database required.

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import (
    create_async_engine, async_sessionmaker, AsyncSession
)

from app.main import create_app
from app.database import get_db_session, sessionmanager
from app.models.base import Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"


@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session_factory = async_sessionmaker(
        bind=engine, class_=AsyncSession, expire_on_commit=False
    )

    async with session_factory() as session:
        yield session

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()


@pytest_asyncio.fixture
async def client(db_session):
    app = create_app()

    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db_session] = override_get_db

    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport, base_url="http://test"
    ) as ac:
        yield ac

# tests/test_users.py
import pytest


@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/api/v1/users/", json={
        "email": "[e-mail]",
        "name": "Test User",
        "password": "securepassword123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[e-mail]"
    assert "id" in data
    assert "password" not in data  # Never expose passwords


@pytest.mark.asyncio
async def test_create_duplicate_user(client):
    user_data = {
        "email": "[e-mail]",
        "name": "First User",
        "password": "password123",
    }
    await client.post("/api/v1/users/", json=user_data)
    response = await client.post("/api/v1/users/", json=user_data)
    assert response.status_code == 409


@pytest.mark.asyncio
async def test_get_nonexistent_user(client):
    response = await client.get("/api/v1/users/99999")
    assert response.status_code == 404

Docker Deployment

# Dockerfile
FROM python:3.12-slim AS base

WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Install dependencies
COPY pyproject.toml .
RUN pip install --no-cache-dir -e ".[prod]"

# Copy application
COPY app/ app/
COPY migrations/ migrations/
COPY alembic.ini .

# Non-root user
RUN adduser --disabled-password --no-create-home appuser
USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml
services:
  app:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: myapp
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U myapp"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  pgdata:

Key Principles Recap

1. Routes are thin — They call services, that's it.
2. Services own business logic — Testable without HTTP context.
3. Repositories own data access — Swap databases without touching logic.
4. Configuration is typed — Pydantic Settings catches misconfig at startup.
5. Dependencies are injected — Easy to override in tests.
6. Errors are consistent — Global exception handlers, structured responses.

This architecture handles everything from a 5-endpoint MVP to a 200-endpoint enterprise API. The investment in structure pays for itself the first time you need to add a feature, fix a bug, or onboard a new teammate.

---

If you found this useful and want this exact architecture pre-built and ready to deploy, check out DataStack Pro — it includes production-tested Python project templates for FastAPI, background workers, and data pipelines, so you can skip the boilerplate and ship faster.