FastAPI SQLAlchemy Integration

Introduction

When building web applications with FastAPI, you'll often need to interact with a database to store and retrieve data. While FastAPI doesn't come with a built-in ORM (Object-Relational Mapper), it integrates seamlessly with SQLAlchemy, one of Python's most popular and powerful ORM libraries.

In this guide, we'll explore how to:

• Set up SQLAlchemy with FastAPI
• Define database models
• Perform CRUD operations (Create, Read, Update, Delete)
• Implement best practices for database connections

By the end of this tutorial, you'll have a solid foundation for building database-powered FastAPI applications.

Prerequisites

Before starting, make sure you have:

• Basic knowledge of Python
• Understanding of FastAPI fundamentals
• A Python environment with FastAPI installed

Let's install the necessary packages:

bash

pip install fastapi sqlalchemy alembic psycopg2-binary uvicorn

• fastapi: Our web framework
• sqlalchemy: The ORM we'll use
• alembic: For database migrations (optional but recommended)
• psycopg2-binary: PostgreSQL driver (you might use a different one based on your database)
• uvicorn: ASGI server to run our FastAPI application

Setting Up SQLAlchemy with FastAPI

Step 1: Database Configuration

First, let's create a structure for our application:

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── schemas.py
└── requirements.txt

Now, let's set up our database connection in database.py:

python

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Change this to your database URL
SQLALCHEMY_DATABASE_URL = "postgresql://username:password@localhost/db_name"

# For SQLite, you would use:
# SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

# Create SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Create a Base class for models
Base = declarative_base()

# Dependency to get DB session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 2: Define Models

Next, let's create our database models in models.py:

python

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    # Define relationship with Item model
    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"))

    # Define relationship with User model
    owner = relationship("User", back_populates="items")

Step 3: Create Pydantic Models (Schemas)

In schemas.py, we'll define Pydantic models for data validation and serialization:

python

from typing import List, Optional
from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    # Enable ORM mode
    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    # Enable ORM mode
    class Config:
        orm_mode = True

Step 4: Create the FastAPI Application

Now, let's create the FastAPI application in main.py:

python

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import models, schemas
from .database import get_db, engine

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

app = FastAPI()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # In a real app, hash the password
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

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

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(models.Item).offset(skip).limit(limit).all()
    return items

Running the Application

To run your application, execute:

bash

uvicorn app.main:app --reload

Testing the Endpoints

Creating a User

Request:

http

POST /users/
Content-Type: application/json

{
  "email": "[email protected]",
  "password": "password123"
}

Response:

json

{
  "id": 1,
  "email": "[email protected]",
  "is_active": true,
  "items": []
}

Creating an Item for a User

Request:

http

POST /users/1/items/
Content-Type: application/json

{
  "title": "My first item",
  "description": "This is an example item"
}

Response:

json

{
  "id": 1,
  "title": "My first item",
  "description": "This is an example item",
  "owner_id": 1
}

Best Practices

1. Use Dependency Injection

FastAPI's dependency injection system is perfect for managing database sessions:

python

def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/{user_id}")
def read_user(user: models.User = Depends(get_user)):
    return user

2. Organize Large Applications

For larger applications, consider organizing your code by feature:

app/
├── api/
│   ├── __init__.py
│   ├── dependencies.py
│   └── endpoints/
│       ├── __init__.py
│       ├── users.py
│       └── items.py
├── core/
│   ├── __init__.py
│   └── config.py
├── db/
│   ├── __init__.py
│   ├── base.py
│   └── session.py
├── models/
│   ├── __init__.py
│   ├── user.py
│   └── item.py
└── schemas/
    ├── __init__.py
    ├── user.py
    └── item.py

3. Use Alembic for Migrations

As your database schema evolves, use Alembic to manage migrations:

bash

# Initialize Alembic
alembic init alembic

# Create a migration
alembic revision --autogenerate -m "Create users and items tables"

# Apply the migration
alembic upgrade head

4. Implement Pagination

For endpoints that return multiple items, always implement pagination:

python

@app.get("/users/", response_model=list[schemas.User])
def read_users(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

Real-World Example: Blog API

Let's create a simple blog API to demonstrate a more complete example:

Models

python

# models.py
from sqlalchemy import Column, ForeignKey, Integer, String, Text, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from .database import Base

class Author(Base):
    __tablename__ = "authors"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    bio = Column(Text, nullable=True)
    
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(Text)
    published = Column(DateTime(timezone=True), server_default=func.now())
    author_id = Column(Integer, ForeignKey("authors.id"))
    
    author = relationship("Author", back_populates="posts")
    comments = relationship("Comment", back_populates="post")

class Comment(Base):
    __tablename__ = "comments"

    id = Column(Integer, primary_key=True, index=True)
    content = Column(Text)
    published = Column(DateTime(timezone=True), server_default=func.now())
    post_id = Column(Integer, ForeignKey("posts.id"))
    
    post = relationship("Post", back_populates="comments")

Schemas

python

# schemas.py
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, EmailStr

# Comment schemas
class CommentBase(BaseModel):
    content: str

class CommentCreate(CommentBase):
    pass

class Comment(CommentBase):
    id: int
    published: datetime
    post_id: int

    class Config:
        orm_mode = True

# Post schemas
class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    published: datetime
    author_id: int
    comments: List[Comment] = []

    class Config:
        orm_mode = True

# Author schemas
class AuthorBase(BaseModel):
    name: str
    email: EmailStr
    bio: Optional[str] = None

class AuthorCreate(AuthorBase):
    pass

class Author(AuthorBase):
    id: int
    posts: List[Post] = []

    class Config:
        orm_mode = True

API Routes

python

# main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from typing import List

from . import models, schemas
from .database import get_db, engine

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

app = FastAPI(title="Blog API")

# Author routes
@app.post("/authors/", response_model=schemas.Author)
def create_author(author: schemas.AuthorCreate, db: Session = Depends(get_db)):
    db_author = models.Author(**author.dict())
    db.add(db_author)
    db.commit()
    db.refresh(db_author)
    return db_author

@app.get("/authors/", response_model=List[schemas.Author])
def read_authors(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    authors = db.query(models.Author).offset(skip).limit(limit).all()
    return authors

@app.get("/authors/{author_id}", response_model=schemas.Author)
def read_author(author_id: int, db: Session = Depends(get_db)):
    db_author = db.query(models.Author).filter(models.Author.id == author_id).first()
    if db_author is None:
        raise HTTPException(status_code=404, detail="Author not found")
    return db_author

# Post routes
@app.post("/authors/{author_id}/posts/", response_model=schemas.Post)
def create_post(
    author_id: int, post: schemas.PostCreate, db: Session = Depends(get_db)
):
    db_post = models.Post(**post.dict(), author_id=author_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

@app.get("/posts/", response_model=List[schemas.Post])
def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    posts = db.query(models.Post).offset(skip).limit(limit).all()
    return posts

@app.get("/posts/{post_id}", response_model=schemas.Post)
def read_post(post_id: int, db: Session = Depends(get_db)):
    db_post = db.query(models.Post).filter(models.Post.id == post_id).first()
    if db_post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    return db_post

# Comment routes
@app.post("/posts/{post_id}/comments/", response_model=schemas.Comment)
def create_comment(
    post_id: int, comment: schemas.CommentCreate, db: Session = Depends(get_db)
):
    db_comment = models.Comment(**comment.dict(), post_id=post_id)
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)
    return db_comment

Summary

In this guide, we've covered:

1. Setting up SQLAlchemy with FastAPI
2. Defining database models and Pydantic schemas
3. Creating CRUD operations using FastAPI endpoints
4. Implementing best practices for database operations
5. Building a real-world example of a blog API

SQLAlchemy integration enables FastAPI applications to interact with relational databases efficiently. The combination provides a robust foundation for building high-performance, type-safe APIs with database persistence.

Further Reading and Exercises

Additional Resources

• SQLAlchemy Documentation
• FastAPI Database Tutorial
• Alembic Documentation for database migrations

Exercises

1. Extend the Blog API: Add functionality for categories and tags for blog posts
2. Implement Authentication: Add user authentication using JWT or OAuth2
3. Add Filtering and Sorting: Enhance the list endpoints with filtering and sorting capabilities
4. Implement Soft Delete: Instead of deleting records, mark them as inactive
5. Create Admin Dashboard: Build a simple admin interface to manage blog content

By completing these exercises, you'll gain hands-on experience with FastAPI and SQLAlchemy integration, preparing you for building real-world applications.