FastAPI Database Models

Database models are the backbone of your FastAPI application's persistence layer. They define how your data is structured in the database and how it can be interacted with in your code. In this tutorial, we'll explore how to create and work with database models in FastAPI using SQLAlchemy ORM (Object-Relational Mapping).

Introduction to Database Models

Database models represent tables in your database as Python classes, and rows as instances of those classes. This abstraction allows you to interact with your database using Python code rather than writing raw SQL queries.

With FastAPI, we typically use SQLAlchemy as the ORM of choice due to its robustness and seamless integration with FastAPI's async features.

Prerequisites

Before diving into database models, you should have:

• Python 3.7+ installed
• Basic knowledge of FastAPI
• Basic understanding of databases
• FastAPI and SQLAlchemy installed:

bash

pip install fastapi sqlalchemy

Setting Up the Database Connection

First, let's create a database configuration file (database.py) to establish our database connection:

python

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

# Create a SQLite database URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# For PostgreSQL you'd use: postgresql://user:password@postgresserver/db

# Create the SQLAlchemy engine
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# The connect_args is needed only for SQLite. For other databases, it's not required.

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

# Create a Base class
Base = declarative_base()

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

Creating Database Models

Now, let's create some database models. We'll create a sample application for managing blog posts and users.

Create a new file called models.py:

python

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

from .database import Base

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)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    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(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    author_id = Column(Integer, ForeignKey("users.id"))
    
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")

class Comment(Base):
    __tablename__ = "comments"
    
    id = Column(Integer, primary_key=True, index=True)
    text = Column(Text)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    post_id = Column(Integer, ForeignKey("posts.id"))
    user_id = Column(Integer, ForeignKey("users.id"))
    
    post = relationship("Post", back_populates="comments")
    user = relationship("User")

Let's break down what we've done:

1. Table Definition: The __tablename__ attribute specifies the name of the table in the database.

2. Columns: We define columns using the Column class, specifying the data type and constraints:

   • Integer, String, Text, Boolean, DateTime are data types
   • primary_key=True marks a column as the primary key
   • index=True creates an index for faster queries
   • unique=True ensures column values are unique
   • ForeignKey establishes relationships between tables

3. Relationships: The relationship function creates relationships between tables:

   • back_populates links relationships between models
   • cascade specifies how operations should propagate to related records

Creating the Database Tables

Now, we need to create these tables in our database. In your main application file:

python

from fastapi import FastAPI
from .database import engine
from . import models

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

app = FastAPI()

# Your API endpoints will go here

Pydantic Models for Data Validation

While SQLAlchemy models define the database structure, Pydantic models define the structure of the data that is sent to and from the API. Let's create a schemas.py file:

python

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

# User schemas
class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    
    class Config:
        orm_mode = True

# Post schemas
class PostBase(BaseModel):
    title: str
    content: str
    published: bool = False

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    created_at: datetime
    updated_at: Optional[datetime] = None
    author_id: int
    
    class Config:
        orm_mode = True

class PostWithUser(Post):
    author: User
    
    class Config:
        orm_mode = True

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

class CommentCreate(CommentBase):
    pass

class Comment(CommentBase):
    id: int
    created_at: datetime
    post_id: int
    user_id: int
    
    class Config:
        orm_mode = True

Notice the Config class with orm_mode = True. This allows Pydantic to work with ORM objects, converting them to dictionaries automatically.

Using Models in API Endpoints

Now, let's create some API endpoints that use our models:

python

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

from . import models, schemas
from .database import get_db

app = FastAPI()

# Create a user
@app.post("/users/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Check if user with this email already exists
    db_user_email = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user_email:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # Check if username is taken
    db_user_name = db.query(models.User).filter(models.User.username == user.username).first()
    if db_user_name:
        raise HTTPException(status_code=400, detail="Username already taken")
    
    # Create a hashed password (in a real app, use proper hashing)
    fake_hashed_password = user.password + "_hashedsecurely"
    
    # Create a new User object
    db_user = models.User(
        email=user.email,
        username=user.username,
        hashed_password=fake_hashed_password
    )
    
    # Add to database, commit changes, and refresh to get generated id
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    return db_user

# Get all users
@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

# Create a post for a user
@app.post("/users/{user_id}/posts/", response_model=schemas.Post, status_code=status.HTTP_201_CREATED)
def create_post(
    user_id: int, 
    post: schemas.PostCreate, 
    db: Session = Depends(get_db)
):
    # Check if user exists
    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")
    
    # Create new post with user_id
    db_post = models.Post(**post.dict(), author_id=user_id)
    
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    
    return db_post

# Get all posts
@app.get("/posts/", response_model=List[schemas.PostWithUser])
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

In the example above:

1. We use Depends(get_db) to inject a database session into our endpoint functions.
2. We use SQLAlchemy queries to interact with our database models.
3. We use Pydantic schemas for request validation and response modeling.

Database Model Relationships

Our examples demonstrate several types of relationships:

1. One-to-Many: A User has many Posts.

   python

   # In User model
   posts = relationship("Post", back_populates="author")
   
   # In Post model
   author_id = Column(Integer, ForeignKey("users.id"))
   author = relationship("User", back_populates="posts")

2. One-to-Many: A Post has many Comments.

   python

   # In Post model
   comments = relationship("Comment", back_populates="post")
   
   # In Comment model
   post_id = Column(Integer, ForeignKey("posts.id"))
   post = relationship("Post", back_populates="comments")

Advanced Model Features

Cascading Deletes

Notice the cascade="all, delete-orphan" parameter in the Post-Comment relationship. This ensures that when a post is deleted, all its comments are deleted as well:

python

comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")

Automatic Timestamps

We've used SQLAlchemy's func.now() to automatically set creation and update timestamps:

python

created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())

Indexes

Indexes speed up database queries. We've added them to commonly queried fields:

python

email = Column(String, unique=True, index=True)
username = Column(String, unique=True, index=True)

Real-World Example: Complete Blog API

Let's expand our example to include more API endpoints for a complete blog application:

python

# Add a comment to a post
@app.post("/posts/{post_id}/comments/", response_model=schemas.Comment)
def create_comment(
    post_id: int, 
    user_id: int, 
    comment: schemas.CommentCreate, 
    db: Session = Depends(get_db)
):
    # Check if post exists
    post = db.query(models.Post).filter(models.Post.id == post_id).first()
    if post is None:
        raise HTTPException(status_code=404, detail="Post not found")
    
    # Check if user exists
    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")
    
    # Create comment
    db_comment = models.Comment(
        **comment.dict(), post_id=post_id, user_id=user_id
    )
    
    db.add(db_comment)
    db.commit()
    db.refresh(db_comment)
    
    return db_comment

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

# Update a post
@app.put("/posts/{post_id}", response_model=schemas.Post)
def update_post(
    post_id: int, 
    post_update: schemas.PostCreate, 
    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")
    
    # Update post attributes
    for key, value in post_update.dict().items():
        setattr(db_post, key, value)
    
    db.commit()
    db.refresh(db_post)
    return db_post

# Delete a post
@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_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")
    
    db.delete(db_post)
    db.commit()
    return None

Best Practices for Database Models

1. Separate Concerns: Keep your database models, Pydantic schemas, and business logic separate.

2. Use Meaningful Names: Use descriptive names for models, columns, and relationships.

3. Add Proper Constraints: Use appropriate constraints like unique=True, nullable=False, etc.

4. Include Indexes: Add indexes to columns that are frequently queried.

5. Use Relationships Properly: Define relationships with appropriate cascade options.

6. Include Audit Trails: Add created and updated timestamps to track changes.

7. Handle Exceptions: Always handle database exceptions gracefully.

Summary

Database models are a crucial part of any FastAPI application that uses a database. In this guide, we've learned:

• How to define SQLAlchemy models for our database tables
• How to establish relationships between models
• How to use these models in FastAPI API endpoints
• Best practices for designing database models

With these skills, you can now build robust FastAPI applications that interact efficiently with databases.

Further Resources and Exercises

Resources

• SQLAlchemy Documentation
• FastAPI Database Documentation
• Pydantic Documentation

Exercises

1. Extend the Blog API: Add categories and tags to posts.

2. Implement Authentication: Add JWT authentication to the API.

3. Add Pagination: Implement proper pagination for listing endpoints.

4. Create Search Functionality: Implement a search endpoint for posts.

5. Add Soft Delete: Implement soft delete functionality instead of permanently removing records.

By completing these exercises, you'll gain a deeper understanding of working with database models in FastAPI applications.