FastAPI Code Organization

Introduction

As your FastAPI application grows, proper code organization becomes crucial. A well-organized project structure makes your code easier to maintain, collaborate on, and scale. In this guide, we'll explore best practices for organizing FastAPI projects, from simple applications to complex microservices.

Proper code organization helps with:

• Improved readability and maintainability
• Better separation of concerns
• Easier testing
• Simplified onboarding for new developers
• More sustainable long-term development

Let's dive into how to structure your FastAPI applications effectively!

Basic Project Structure

For small to medium-sized applications, a straightforward but well-organized structure works best. Here's a recommended basic project layout:

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app creation and API routes
│   ├── models/           # Pydantic models
│   │   ├── __init__.py
│   │   └── user.py       # e.g., User model
│   ├── routers/          # API route modules
│   │   ├── __init__.py
│   │   └── users.py      # e.g., user-related routes
│   ├── services/         # Business logic
│   │   ├── __init__.py
│   │   └── user.py       # e.g., user service functions
│   └── database.py       # Database connection and models
├── tests/                # Test files
│   ├── __init__.py
│   └── test_users.py
├── .env                  # Environment variables (add to .gitignore)
├── .gitignore
├── requirements.txt
└── README.md

Let's understand the purpose of each component:

• main.py: Entry point for your application, creates the FastAPI instance
• models/: Pydantic models for request/response validation
• routers/: API route handlers organized by feature/resource
• services/: Business logic isolated from the API layer
• database.py: Database connection and ORM model definitions

Creating the Application Entry Point

The entry point (main.py) initializes your FastAPI application and includes routers:

python

from fastapi import FastAPI
from app.routers import users

app = FastAPI(
    title="My FastAPI App",
    description="A well-organized FastAPI application",
    version="0.1.0"
)

# Include routers
app.include_router(users.router)

@app.get("/")
async def root():
    return {"message": "Welcome to my FastAPI application!"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

Defining Pydantic Models

Create clear Pydantic models in the models directory. For example, in app/models/user.py:

python

from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)
    full_name: Optional[str] = None
    
class UserCreate(UserBase):
    password: str = Field(..., min_length=8)
    
class UserResponse(UserBase):
    id: int
    created_at: datetime
    
    class Config:
        orm_mode = True

Organizing Routers

Separate your routes by resource or feature in the routers directory. For example, in app/routers/users.py:

python

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List
from sqlalchemy.orm import Session

from app.models.user import UserCreate, UserResponse
from app.services.user import create_user, get_user_by_id, get_users
from app.database import get_db

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_new_user(user: UserCreate, db: Session = Depends(get_db)):
    return create_user(db=db, user=user)

@router.get("/", response_model=List[UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = get_users(db, skip=skip, limit=limit)
    return users

@router.get("/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = get_user_by_id(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

Implementing Business Logic in Services

Keep your business logic separate from route handlers in the services directory. For example, in app/services/user.py:

python

from sqlalchemy.orm import Session
from fastapi import HTTPException, status
from typing import List, Optional

from app.database import User
from app.models.user import UserCreate

def get_user_by_id(db: Session, user_id: int) -> Optional[User]:
    return db.query(User).filter(User.id == user_id).first()

def get_user_by_email(db: Session, email: str) -> Optional[User]:
    return db.query(User).filter(User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate) -> User:
    # Check if user with this email already exists
    existing_user = get_user_by_email(db, email=user.email)
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )
    
    # Create new user (with hashed password in a real application)
    db_user = User(
        email=user.email,
        username=user.username,
        full_name=user.full_name,
        hashed_password=user.password  # In a real app: get_password_hash(user.password)
    )
    
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Database Connection and Models

Define your database setup in app/database.py:

python

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
import os

# In production, use environment variables for connection details
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")

engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

# SQLAlchemy Models
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)
    full_name = Column(String, nullable=True)
    hashed_password = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships example
    items = relationship("Item", back_populates="owner")

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)
    owner_id = Column(Integer, ForeignKey("users.id"))
    
    # Relationship with User
    owner = relationship("User", back_populates="items")

# Create tables
Base.metadata.create_all(bind=engine)

Advanced Organization for Larger Applications

For larger applications, consider a more modular structure organized by feature:

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py        # Configuration handling
│   │   ├── security.py      # Security utilities
│   │   └── database.py      # Database connection
│   ├── api/
│   │   ├── __init__.py
│   │   ├── api_v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── __init__.py
│   │   │   │   ├── users.py
│   │   │   │   └── items.py
│   │   │   └── api.py       # API v1 router setup
│   │   └── deps.py          # Dependency injection
│   ├── models/              # SQLAlchemy models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/             # Pydantic schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── crud/                # CRUD operations
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── user.py
│   │   └── item.py
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── main.py              # FastAPI app creation
├── alembic/                 # Database migrations
├── tests/
└── ...

This structure enables API versioning and clearer separation of concerns.

Environment Configuration

Instead of hardcoding configuration values, use environment variables and a configuration class:

python

# app/core/config.py
import os
from pydantic import BaseSettings, PostgresDsn

class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "My FastAPI App"
    
    # Security settings
    SECRET_KEY: str = os.getenv("SECRET_KEY", "devkey123")
    ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 60 * 24))
    
    # Database settings
    DATABASE_URL: PostgresDsn = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/app")
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()

Dependency Injection

FastAPI's dependency injection system makes your code more modular and testable:

python

# app/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.orm import Session
from pydantic import ValidationError

from app.core.config import settings
from app.core.database import SessionLocal
from app.models.user import User
from app.schemas.token import TokenPayload

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/login/access-token")

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

def get_current_user(
    db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=["HS256"]
        )
        token_data = TokenPayload(**payload)
    except (JWTError, ValidationError):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Could not validate credentials",
        )
    
    user = db.query(User).filter(User.id == token_data.sub).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )
    return user

API Router Setup

Use APIRouter to organize your endpoints:

python

# app/api/api_v1/api.py
from fastapi import APIRouter
from app.api.api_v1.endpoints import users, items

api_router = APIRouter()
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(items.router, prefix="/items", tags=["items"])

Testing Organization

Keep tests organized by feature, mirroring your app structure:

tests/
├── __init__.py
├── conftest.py           # Fixtures and test configuration
├── api/
│   ├── __init__.py
│   ├── test_users.py
│   └── test_items.py
└── services/
    ├── __init__.py
    └── test_user_service.py

Example of a test file:

python

# tests/api/test_users.py
from fastapi.testclient import TestClient
import pytest

def test_create_user(client: TestClient):
    response = client.post(
        "/api/v1/users/",
        json={
            "email": "[email protected]",
            "password": "password123",
            "username": "testuser",
            "full_name": "Test User"
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[email protected]"
    assert data["username"] == "testuser"
    assert "id" in data
    assert "password" not in data

Real-World Project Example

Let's look at a more realistic example of how to handle multiple related routes and services:

python

# app/main.py
from fastapi import FastAPI
from app.api.api_v1.api import api_router
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# Include routers
app.include_router(api_router, prefix=settings.API_V1_STR)

# Health check endpoint
@app.get("/health")
def health_check():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

Summary

Properly organizing your FastAPI code offers significant benefits for maintainability, readability, and collaboration. The key principles to remember are:

1. Separate concerns: Keep models, routes, and business logic separate
2. Use modular structure: Organize by feature or resource
3. Consider scalability: Choose a structure that can grow with your application
4. Leverage dependency injection: Make your code more testable
5. Implement configuration management: Use environment variables instead of hardcoded values
6. Organize by feature: Group related files together

By following these best practices, you'll build FastAPI applications that are easier to maintain, test, and extend over time.

Additional Resources

• FastAPI Official Documentation
• Full Stack FastAPI and PostgreSQL
• SQLAlchemy Documentation
• Pydantic Documentation

Exercises

1. Create a basic FastAPI project structure with users and posts resources
2. Convert a single-file FastAPI app into a well-structured project
3. Implement dependency injection for database and authentication
4. Add configuration management using environment variables
5. Write tests for the restructured application

Happy coding!