Add a Database to FastAPI with SQLAlchemy Models

How to Add a Database to FastAPI with SQLAlchemy Models

In this tutorial, you will learn how to integrate a database into your FastAPI application using SQLAlchemy. We will cover setting up database models, defining relationships between them, updating Pydantic schemas to work with SQLAlchemy models, and implementing API endpoints for data creation and retrieval. This will replace the temporary in-memory data storage with persistent data storage.

Prerequisites

• Basic knowledge of Python
• Familiarity with FastAPI and Pydantic (from previous tutorials)
• A working FastAPI project setup

Overview of Application Architecture

We will structure our application into three distinct layers to ensure maintainability and clarity:

1. Database Models (SQLAlchemy): These are Python classes that map directly to database tables, defining the structure of our data.
2. Pydantic Schemas: These define the structure of the data that our API accepts as input and returns as output, ensuring data validation.
3. API Routes (FastAPI Endpoints): These are the functions that handle incoming HTTP requests and interact with the database through our models and schemas.

While libraries like SQLModel combine SQLAlchemy and Pydantic, using separate models and schemas offers greater control, better understanding of underlying concepts, and aligns with common industry practices in production FastAPI applications.

Step 1: Install SQLAlchemy

First, you need to install SQLAlchemy. If you are using pip, run:

pip install sqlalchemy

If you are using UV (as in the tutorial), run:

uv add sqlalchemy

Step 2: Set Up Database Configuration

Create a new file named database.py. This file will contain all the necessary code for connecting to and managing our database sessions. For this tutorial, we will use SQLite, which is built into Python and requires no separate installation.

Paste the following code into database.py:

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

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

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

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

Base = declarative_base()

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

Explanation:

• SQLALCHEMY_DATABASE_URL: Specifies the connection string for the database. For SQLite, sqlite:///./blog.db points to a file named blog.db in the current directory.
• engine: Creates a database engine, which is the starting point for SQLAlchemy. check_same_thread=False is specific to SQLite and allows multiple threads to access the database, which is necessary for FastAPI.
• SessionLocal: A factory for creating database sessions. A session represents a transaction with the database.
• Base: A declarative base class from which all our database models will inherit.
• get_db: A dependency function that provides a database session to our API routes. It uses a generator with a yield statement to ensure the session is properly closed after each request, even if errors occur.

Step 3: Define Database Models

Create a new file named models.py. This file will define the structure of our database tables using SQLAlchemy’s Object-Relational Mapper (ORM).

Paste the following code into models.py:

from datetime import datetime
from typing import List, Optional

from sqlalchemy import ForeignKey, text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(unique=True, index=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    image_file: Mapped[Optional[str]] = mapped_column(nullable=True)

    posts: Mapped["Post"] = relationship(back_populates="author")

    @property
    def image_path(self):
        if self.image_file:
            return f"/media/profile_pics/{self.image_file}"
        return f"/static/profile_pics/default.jpeg"

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(index=True)
    content: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    date_posted: Mapped[datetime] = mapped_column(server_default=text("CURRENT_TIMESTAMP"))

    author: Mapped[User] = relationship(back_populates="posts")

Explanation:

• User Model:
  • __tablename__ = "users": Sets the name of the table in the database.
  • id: Primary key, auto-incrementing.
  • username, email: Unique string fields, indexed for faster lookups.
  • image_file: An optional string to store the filename of a user’s profile picture.
  • posts: Mapped["Post"] = relationship(back_populates="author"): Defines a one-to-many relationship, meaning a user can have multiple posts. back_populates links this to the author field in the Post model.
  • @property def image_path(self):: A computed property that generates the full path to the user’s profile image, distinguishing between uploaded images and default images.
• Post Model:
  • __tablename__ = "posts": Sets the table name.
  • id: Primary key, auto-incrementing.
  • title: String field, indexed.
  • content: String field for the post content.
  • user_id: Mapped[int] = mapped_column(ForeignKey("users.id")): A foreign key linking each post to a user’s ID.
  • date_posted: Mapped[datetime] = mapped_column(server_default=text("CURRENT_TIMESTAMP")): A timestamp for when the post was created, defaulting to the current time.
  • author: Mapped[User] = relationship(back_populates="posts"): Defines the many-to-one relationship, linking each post back to its author (a User object).

Note on Forward References: When referencing a model (like Post in User) before it’s defined, Python versions prior to 3.10 might require from __future__ import annotations as the very first import in the file. The provided code works with modern Python versions and includes the import for broader compatibility.

Step 4: Update Pydantic Schemas

Now, update your existing Pydantic schemas (likely in a file like schemas.py) to accommodate the new database models and relationships. You’ll need to define schemas for users and adjust the post schemas.

First, ensure you have the necessary imports in your schemas file:

from datetime import datetime
from typing import Optional

from pydantic import BaseModel, EmailStr, Field

# Import SQLAlchemy models
import models

Define User Schemas:

class UserBase(BaseModel):
    username: str = Field(min_length=1, max_length=50)
    email: EmailStr = Field(max_length=120)

class UserCreate(UserBase):
    pass  # For now, same as UserBase. Will add password later.

class UserResponse(UserBase):
    id: int
    image_file: Optional[str] = None
    image_path: str # This will be a computed property from the model

    class Config:
        from_attributes = True # Allows Pydantic to read from SQLAlchemy models

Update Post Schemas:

class PostBase(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    content: str

class PostCreate(PostBase):
    user_id: int # Temporarily required, will be inferred from session later

class PostResponse(PostBase):
    id: int
    user_id: int
    date_posted: datetime
    author: UserResponse # Nested author information

    class Config:
        from_attributes = True

Explanation:

• User Schemas: UserBase defines common fields. UserCreate is for input (currently without password). UserResponse includes the user’s ID and the computed image_path. Config.from_attributes = True enables reading data directly from SQLAlchemy model instances.
• Post Schemas:
  • PostBase remains similar.
  • PostCreate now includes user_id, which will be manually provided for now.
  • PostResponse is significantly updated: it includes the author field, which is a UserResponse object. This allows for nested JSON output, embedding user details within each post response. Pydantic automatically serializes datetime objects to the ISO 8601 format.

Step 5: Update Main Application File (main.py)

Modify your main FastAPI application file (e.g., main.py) to incorporate the database setup, create tables, mount media directories, and update your API routes.

Add the following imports at the top of main.py:

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

# Import database and models
from database import engine, get_db
import models
import schemas

Initialize the app, create tables, and mount media directories:

app = FastAPI()

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

# Mount media directory for user uploads
from fastapi.staticfiles import StaticFiles
app.mount("/media", StaticFiles(directory="media"), name="media")

# Create media subdirectories if they don't exist
import os
if not os.path.exists("media/profile_pics"): 
    os.makedirs("media/profile_pics")

Important: Remove the in-memory post list that was used in previous tutorials.

Step 6: Implement User Endpoints

Add new routes for creating users. This demonstrates how to use the database session and handle potential conflicts.

Add the following route to main.py:

@app.post("/api/users", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(
    user: schemas.UserCreate,
    db: Annotated[Session, Depends(get_db)]
):
    # Check if username already exists
    existing_user = db.scalar(select(models.User).where(models.User.username == user.username))
    if existing_user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists")

    # Check if email already exists
    existing_email = db.scalar(select(models.User).where(models.User.email == user.email))
    if existing_email:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email already exists")

    # Create new user instance
    new_user = models.User(username=user.username, email=user.email)
    
    # Add and commit to database
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    
    return new_user

Explanation:

• Dependency Injection: db: Annotated[Session, Depends(get_db)] injects a database session into the route function. FastAPI automatically calls get_db and provides the session.
• Unique Constraints Check: Before creating a user, the code queries the database to ensure the username and email are not already in use, returning a 400 Bad Request error if they are.
• Creating and Saving: A new models.User instance is created, added to the session (db.add()), committed to the database (db.commit()), and then refreshed to load any database-generated fields like the ID (db.refresh()).
• Response Model: The route returns a schemas.UserResponse, and Pydantic handles the conversion.

Step 7: Update Post Endpoints

Modify the existing post creation route to use the database instead of the in-memory list. You’ll also need to adjust how posts are retrieved to include related user data.

Update the create_post route in main.py:

@app.post("/api/posts", response_model=schemas.PostResponse, status_code=status.HTTP_201_CREATED)
def create_post(
    post: schemas.PostCreate,
    db: Annotated[Session, Depends(get_db)]
):
    # Create new post instance
    new_post = models.Post(
        title=post.title,
        content=post.content,
        user_id=post.user_id # Use the provided user_id
    )
    
    # Add and commit to database
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    
    # Return the created post, including author details
    return new_post

Add a route to get all posts, including their authors:

@app.get("/api/posts", response_model=list[schemas.PostResponse])
def get_posts(
    db: Annotated[Session, Depends(get_db)]
):
    # Select posts and join with users to get author information
    posts = db.scalars(select(models.Post).options(sqlalchemy.orm.joinedload(models.Post.author)))
    return list(posts)

Explanation:

• Create Post: The create_post function now takes a schemas.PostCreate object and a database session. It creates a models.Post instance using the provided data, including the user_id, and saves it to the database. The response includes the full PostResponse, which automatically embeds the related author data thanks to the relationship setup and Pydantic’s configuration.
• Get Posts: The get_posts route queries the database for all posts. select(models.Post).options(sqlalchemy.orm.joinedload(models.Post.author)) is used to efficiently load the related author data for each post in a single query using a JOIN. The result is returned as a list of schemas.PostResponse objects.

By following these steps, you have successfully integrated a database with SQLAlchemy into your FastAPI application, enabling persistent storage and defining relationships between your data models.

---

Source: Python FastAPI Tutorial (Part 5): Adding a Database – SQLAlchemy Models and Relationships (YouTube)