FastAPI Database Integration with SQLAlchemy

FastAPI doesn’t require you to use SQL databases, but you can integrate them using SQLAlchemy, a Python SQL toolkit and Object Relational Mapper. This article covers setting up SQLAlchemy with FastAPI and implementing CRUD operations.

Understanding SQLAlchemy

SQLAlchemy is a Python SQL toolkit and Object Relational Mapper. An ORM maps your code objects to database tables, allowing you to work with databases using Python objects instead of writing raw SQL.

For example, a Python class Blog can represent a SQL table blogs when you set __tablename__ = 'blogs'. This abstraction makes database operations more intuitive and maintainable.

Project Structure

Organize your FastAPI application with a clear structure. Create a blog directory and make it a Python module by adding __init__.py:

blog/
  __init__.py
  main.py
  schemas.py
  database.py
  models.py

Setting Up Database Connection

Create database.py to configure the database connection:

# blog/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = 'sqlite:///./blog.db'

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}  # needed for SQLite with threads
)

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

The engine manages database connections. SessionLocal creates database sessions. Base is used for declarative model definitions. The connect_args parameter is specific to SQLite and allows it to work with FastAPI’s async nature.

Creating Models

Create models.py to define your database models:

# blog/models.py
from sqlalchemy import Column, Integer, String
from .database import Base

class Blog(Base):
    __tablename__ = 'blogs'
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    body = Column(String)

Models extend the Base class and define table structure using SQLAlchemy Column types. The __tablename__ attribute specifies the table name in the database.

Creating Tables

In your main.py, create tables from your models:

# blog/main.py
from fastapi import FastAPI
from . import models
from .database import engine

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

create_all creates tables if they don’t exist. This is not a migrations tool; it’s useful for development but won’t alter existing schemas. For production, use proper migration tools like Alembic.

Database Sessions

To use the database in your route handlers, you need database sessions. Create a dependency function that provides a session:

# blog/main.py
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from .database import SessionLocal, engine
from . import models

app = FastAPI()
models.Base.metadata.create_all(bind=engine)

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

Using Depends(get_db) is the recommended way. When the request ends, the finally block closes the session, ensuring proper cleanup.

Creating Records

Implement a POST endpoint to create records:

from fastapi import Depends, status
from sqlalchemy.orm import Session
from . import schemas, models
from .database import get_db

@app.post('/blog', status_code=status.HTTP_201_CREATED)
def create(request: schemas.Blog, db: Session = Depends(get_db)):
    new_blog = models.Blog(title=request.title, body=request.body)
    db.add(new_blog)
    db.commit()
    db.refresh(new_blog)
    return new_blog

The process is: create a model instance, add it to the session, commit the transaction, refresh to get database-generated values like ID, and return the created object.

Reading Records

To get all records:

@app.get('/blog')
def all(db: Session = Depends(get_db)):
    blogs = db.query(models.Blog).all()
    return blogs

To get a specific record by ID:

@app.get('/blog/{id}', status_code=status.HTTP_200_OK)
def show(id: int, db: Session = Depends(get_db)):
    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
    return blog

Error Handling

Handle cases where resources don’t exist:

from fastapi import HTTPException, status

@app.get('/blog/{id}', status_code=status.HTTP_200_OK)
def show(id: int, db: Session = Depends(get_db)):
    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
    if not blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Blog with id {id} not found"
        )
    return blog

Use raise HTTPException rather than returning it. FastAPI will convert the exception into a proper HTTP error response.

Updating Records

Implement an update endpoint:

@app.put('/blog/{id}', status_code=status.HTTP_202_ACCEPTED)
def update(id: int, request: schemas.Blog, db: Session = Depends(get_db)):
    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
    if not blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Blog with id {id} not found"
        )
    blog.title = request.title
    blog.body = request.body
    db.commit()
    db.refresh(blog)
    return blog

Update the model attributes, commit the changes, refresh to get updated values, and return the updated object.

Deleting Records

Implement a delete endpoint:

@app.delete('/blog/{id}', status_code=status.HTTP_204_NO_CONTENT)
def destroy(id: int, db: Session = Depends(get_db)):
    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
    if not blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Blog with id {id} not found"
        )
    db.delete(blog)
    db.commit()

Remember to commit after deleting. The 204 status code indicates successful deletion with no content to return.

Schemas vs Models

Separate Pydantic schemas from SQLAlchemy models. Schemas define request and response structures, while models define database tables:

# blog/schemas.py
from pydantic import BaseModel

class Blog(BaseModel):
    title: str
    body: str

This separation allows you to have different schemas for requests, responses, and database models, providing flexibility and better API design.

Summary

SQLAlchemy integration with FastAPI provides a powerful way to work with databases. By using ORM models, dependency injection for database sessions, and proper error handling, you can build robust database-backed APIs.

The key concepts are understanding the relationship between models and tables, using dependency injection for sessions, implementing proper CRUD operations, and handling errors appropriately. FastAPI’s integration with SQLAlchemy makes database operations clean and maintainable.