How to Structure a Production FastAPI Project

Prerequisites

  • Python 3.11+
  • FastAPI and SQLAlchemy installed
  • Basic understanding of REST APIs
project/
├── app/
│   ├── __init__.py
│   ├── main.py              # App factory, middleware, startup events
│   ├── config.py             # Pydantic Settings
│   ├── dependencies.py       # Shared dependencies (DB session, auth)
│   │
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py     # Combines all v1 routers
│   │   │   ├── users.py      # User endpoints
│   │   │   └── orders.py     # Order endpoints
│   │   └── v2/
│   │       └── router.py
│   │
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py           # SQLAlchemy ORM models
│   │   └── order.py
│   │
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py           # Pydantic request/response schemas
│   │   └── order.py
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py   # Business logic
│   │   └── order_service.py
│   │
│   └── core/
│       ├── __init__.py
│       ├── database.py       # Engine, session factory
│       └── security.py       # Password hashing, JWT helpers
│
├── tests/
│   ├── conftest.py           # Fixtures
│   ├── test_users.py
│   └── test_orders.py
│
├── alembic/                  # Database migrations
├── requirements.txt
├── .env
└── Dockerfile

Key Layers Explained

Models (SQLAlchemy ORM)

# app/models/user.py
from sqlalchemy import Column, Integer, String
from app.core.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String)

Schemas (Pydantic)

# app/schemas/user.py
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str | None = None

class UserResponse(BaseModel):
    id: int
    email: str
    full_name: str | None

    class Config:
        from_attributes = True  # Enables ORM → Pydantic conversion

Services (Business Logic)

# app/services/user_service.py
from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate
from app.core.security import hash_password

class UserService:
    def __init__(self, db: Session):
        self.db = db

    def create_user(self, data: UserCreate) -> User:
        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            full_name=data.full_name
        )
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user

    def get_by_email(self, email: str) -> User | None:
        return self.db.query(User).filter(User.email == email).first()

Dependencies (DI)

# app/dependencies.py
from app.core.database import SessionLocal
from app.core.security import decode_access_token
from fastapi import Depends, HTTPException

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(token: str = Depends(oauth2_scheme)):
    user_id = decode_access_token(token)
    if not user_id:
        raise HTTPException(401, "Invalid token")
    return user_id

Routers (Thin Endpoints)

# app/api/v1/users.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.dependencies import get_db

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

@router.post("/", response_model=UserResponse, status_code=201)
def create_user(data: UserCreate, db: Session = Depends(get_db)):
    service = UserService(db)
    if service.get_by_email(data.email):
        raise HTTPException(400, "Email already exists")
    return service.create_user(data)

Configuration with Pydantic Settings

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql://user:pass@localhost/db"
    secret_key: str
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

Testing

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import create_app
from app.core.database import get_test_db

@pytest.fixture
def client():
    app = create_app()
    app.dependency_overrides[get_db] = get_test_db
    return TestClient(app)

# tests/test_users.py
def test_create_user(client):
    response = client.post("/api/v1/users/", json={
        "email": "[email protected]",
        "password": "secret123"
    })
    assert response.status_code == 201
    assert response.json()["email"] == "[email protected]"

Why This Structure?

  • Routes are thin — they only handle HTTP concerns (parsing, status codes, dependency injection)
  • Services contain business logic — reusable across endpoints and background jobs
  • Schemas decouple API from DB — change your ORM without breaking API contracts
  • Dependencies are explicit — every endpoint declares what it needs
  • Testing is straightforward — mock dependencies at the FastAPI level with dependency_overrides

Summary

  • Separate models (ORM), schemas (API contract), and services (logic)
  • Keep endpoints thin — delegate to service classes
  • Use Pydantic Settings for configuration
  • Organise routes by version: api/v1/, api/v2/
  • Override dependencies in tests with dependency_overrides

References

Advertisement