Skip to main content

FastAPI Database Patterns

In building FastAPI applications that interact with databases, certain patterns emerge that help maintain clean, testable, and maintainable code. This guide explores common database patterns that make your FastAPI applications more robust and easier to extend.

Introduction​

When building APIs with FastAPI, database interactions often form a critical part of your application. How you structure this code can greatly impact:

  • Maintainability
  • Testability
  • Performance
  • Scalability

In this guide, we'll explore several proven patterns for organizing database code in FastAPI applications. These patterns work with any database technology, though our examples will use SQLAlchemy with PostgreSQL as it's a common choice in the FastAPI ecosystem.

Dependency Injection for Database Access​

FastAPI's dependency injection system provides an elegant way to handle database connections.

The Connection Dependency Pattern​

python
from fastapi import Depends, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# Database setup
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

app = FastAPI()

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

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

This pattern offers several benefits:

  • Database sessions are automatically opened and closed
  • The database connection is only active when needed
  • Dependencies can be easily replaced for testing

The Repository Pattern​

The Repository pattern separates data access logic from business logic by creating a layer of abstraction.

Basic Repository Implementation​

python
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from typing import List, Optional

# Models definition omitted for brevity

class UserRepository:
    def __init__(self, db: Session):
        self.db = db
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.db.query(User).filter(User.id == user_id).first()
    
    def get_all(self) -> List[User]:
        return self.db.query(User).all()
    
    def create(self, user_data: dict) -> User:
        user = User(**user_data)
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user
    
    def update(self, user_id: int, user_data: dict) -> Optional[User]:
        user = self.get_by_id(user_id)
        if user:
            for key, value in user_data.items():
                setattr(user, key, value)
            self.db.commit()
            self.db.refresh(user)
        return user
    
    def delete(self, user_id: int) -> bool:
        user = self.get_by_id(user_id)
        if user:
            self.db.delete(user)
            self.db.commit()
            return True
        return False

# In FastAPI routes
def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

@app.get("/users/{user_id}")
def read_user(
    user_id: int, 
    user_repo: UserRepository = Depends(get_user_repository)
):
    user = user_repo.get_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Benefits of the Repository pattern:

  • Separates database logic from API logic
  • Makes testing easier with mock repositories
  • Creates a consistent interface for data operations
  • Centralizes database queries for reuse

Service Layer Pattern​

For complex applications, adding a service layer between repositories and routes helps manage business logic.

python
from fastapi import Depends, FastAPI
from typing import List

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def get_user(self, user_id: int) -> dict:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        
        # Add business logic here
        return {
            "id": user.id,
            "username": user.username,
            "is_active": user.is_active,
            "posts_count": len(user.posts)
        }
    
    def register_user(self, username: str, password: str) -> dict:
        # Check if user exists
        if self.user_repository.get_by_username(username):
            raise HTTPException(400, detail="Username already registered")
        
        # Hash password (business logic)
        hashed_password = hash_password(password)
        
        # Create user
        user = self.user_repository.create({
            "username": username,
            "password": hashed_password,
            "is_active": True
        })
        
        return {"id": user.id, "username": user.username}

# Dependency
def get_user_service(
    user_repo: UserRepository = Depends(get_user_repository)
) -> UserService:
    return UserService(user_repo)

@app.post("/users/", status_code=201)
def create_user(
    user_data: UserCreate,
    user_service: UserService = Depends(get_user_service)
):
    return user_service.register_user(
        username=user_data.username,
        password=user_data.password
    )

Async Database Patterns​

FastAPI excels with asynchronous code. Here's how to use async patterns with databases.

Async SQLAlchemy with FastAPI​

python
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# Async database setup
SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

app = FastAPI()

# Async dependency
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    # Using SQLAlchemy 1.4+ async execution
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalars().first()
    
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Async Repository Pattern​

python
class AsyncUserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_by_id(self, user_id: int) -> Optional[User]:
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalars().first()
    
    async def create(self, user_data: dict) -> User:
        user = User(**user_data)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

# In FastAPI routes
async def get_async_user_repository(
    db: AsyncSession = Depends(get_db)
) -> AsyncUserRepository:
    return AsyncUserRepository(db)

@app.get("/users/{user_id}")
async def read_user(
    user_id: int, 
    user_repo: AsyncUserRepository = Depends(get_async_user_repository)
):
    user = await user_repo.get_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Unit of Work Pattern​

The Unit of Work pattern manages transaction boundaries and ensures all operations within a transaction succeed or fail as one unit.

python
class UnitOfWork:
    def __init__(self, db: Session):
        self.db = db
        self.users = UserRepository(db)
        self.posts = PostRepository(db)
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.rollback()
        else:
            self.commit()
    
    def commit(self):
        self.db.commit()
    
    def rollback(self):
        self.db.rollback()

# FastAPI dependency
def get_uow(db: Session = Depends(get_db)) -> UnitOfWork:
    return UnitOfWork(db)

@app.post("/users/{user_id}/posts/")
def create_post_for_user(
    user_id: int,
    post_data: PostCreate,
    uow: UnitOfWork = Depends(get_uow)
):
    # All operations succeed or fail together
    user = uow.users.get_by_id(user_id)
    if not user:
        raise HTTPException(404, "User not found")
        
    post = uow.posts.create({
        "title": post_data.title,
        "content": post_data.content,
        "user_id": user_id
    })
    
    # No need to commit explicitly, handled by UnitOfWork
    return post

Real-World Example: Blog API​

Let's bring these patterns together in a real-world example of a blog API:

python
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

app = FastAPI()

# === Repositories ===
class PostRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_all(self, skip: int = 0, limit: int = 100):
        result = await self.db.execute(
            select(Post).offset(skip).limit(limit)
        )
        return result.scalars().all()
    
    async def get_by_id(self, post_id: int):
        result = await self.db.execute(
            select(Post).where(Post.id == post_id)
        )
        return result.scalars().first()
    
    async def create(self, post_data: dict):
        post = Post(**post_data)
        self.db.add(post)
        await self.db.commit()
        await self.db.refresh(post)
        return post

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_by_id(self, user_id: int):
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalars().first()

# === Services ===
class BlogService:
    def __init__(
        self,
        post_repository: PostRepository,
        user_repository: UserRepository
    ):
        self.post_repository = post_repository
        self.user_repository = user_repository
    
    async def get_all_posts(self, skip: int = 0, limit: int = 100):
        return await self.post_repository.get_all(skip, limit)
    
    async def get_post(self, post_id: int):
        post = await self.post_repository.get_by_id(post_id)
        if not post:
            raise HTTPException(404, "Post not found")
        return post
    
    async def create_post(self, user_id: int, title: str, content: str):
        # Verify user exists
        user = await self.user_repository.get_by_id(user_id)
        if not user:
            raise HTTPException(404, "User not found")
        
        # Create post
        post_data = {
            "title": title,
            "content": content,
            "user_id": user_id
        }
        
        return await self.post_repository.create(post_data)

# === Dependencies ===
async def get_post_repository(db: AsyncSession = Depends(get_db)):
    return PostRepository(db)

async def get_user_repository(db: AsyncSession = Depends(get_db)):
    return UserRepository(db)

async def get_blog_service(
    post_repo: PostRepository = Depends(get_post_repository),
    user_repo: UserRepository = Depends(get_user_repository)
):
    return BlogService(post_repo, user_repo)

# === Routes ===
@app.get("/posts/", response_model=List[PostRead])
async def read_posts(
    skip: int = 0,
    limit: int = 100,
    blog_service: BlogService = Depends(get_blog_service)
):
    return await blog_service.get_all_posts(skip, limit)

@app.get("/posts/{post_id}", response_model=PostRead)
async def read_post(
    post_id: int,
    blog_service: BlogService = Depends(get_blog_service)
):
    return await blog_service.get_post(post_id)

@app.post("/posts/", response_model=PostRead, status_code=201)
async def create_post(
    post_data: PostCreate,
    blog_service: BlogService = Depends(get_blog_service)
):
    return await blog_service.create_post(
        user_id=post_data.user_id,
        title=post_data.title,
        content=post_data.content
    )

Summary​

Implementing proper database patterns in your FastAPI applications brings numerous benefits:

  • Separation of concerns: Each component has a single responsibility
  • Improved testability: Easily mock dependencies for unit tests
  • Maintainability: Changes to database logic don't affect API endpoints
  • Scalability: Consistent patterns make adding new features easier

The key patterns we've covered:

  1. Dependency Injection for database sessions
  2. Repository Pattern for abstracting data access
  3. Service Layer for business logic
  4. Async patterns for non-blocking database operations
  5. Unit of Work pattern for transaction management

By applying these patterns, you'll build FastAPI applications that are more robust, easier to test, and simpler to maintain as they grow in complexity.

Additional Resources​

Exercises​

  1. Implement a complete CRUD repository for a Product entity with SQLAlchemy
  2. Convert a synchronous repository to use async SQLAlchemy
  3. Implement the Unit of Work pattern for a multi-entity transaction
  4. Create a service layer that uses multiple repositories and implements business rules
  5. Write unit tests for your repositories using an in-memory SQLite database

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)