Skip to main content

FastAPI ORM Relationships

When building real-world applications with FastAPI, you'll often need to model complex data structures with relationships between different entities. Understanding how to properly define and work with these database relationships is crucial for creating efficient, scalable applications.

In this tutorial, we'll explore how to implement various types of relationships using FastAPI and SQLAlchemy ORM.

Prerequisites​

Before we dive into relationships, make sure you have:

  • Basic knowledge of FastAPI
  • Understanding of SQLAlchemy ORM basics
  • A working FastAPI project with SQLAlchemy setup

Types of Database Relationships​

In relational databases, there are three fundamental types of relationships:

  1. One-to-One: When one record in table A is related to exactly one record in table B
  2. One-to-Many: When one record in table A is related to multiple records in table B
  3. Many-to-Many: When multiple records in table A can be related to multiple records in table B

Let's implement each type using SQLAlchemy in our FastAPI application.

Setting Up Base Models​

First, let's set up our SQLAlchemy configuration:

python
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, Session

# Database URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./relationships.db"

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

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

# Base class for models
Base = declarative_base()

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

app = FastAPI()

One-to-One Relationship​

A one-to-one relationship is when one record in a table corresponds to exactly one record in another table. Let's implement a relationship between a User and their Profile.

python
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    
    # Relationship - One-to-One
    profile = relationship("Profile", back_populates="user", uselist=False)

class Profile(Base):
    __tablename__ = "profiles"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), unique=True)
    full_name = Column(String, index=True)
    bio = Column(String)
    
    # Relationship - One-to-One
    user = relationship("User", back_populates="profile")

Key points about this implementation:

  • uselist=False tells SQLAlchemy that the relationship is one-to-one (not a collection)
  • The ForeignKey in the Profile model points to the id field of the users table
  • unique=True for the user_id ensures that each user can have only one profile

Let's create API endpoints to work with this relationship:

python
from pydantic import BaseModel
from typing import Optional

# Pydantic models for request/response
class ProfileBase(BaseModel):
    full_name: str
    bio: Optional[str] = None

class ProfileCreate(ProfileBase):
    pass

class Profile(ProfileBase):
    id: int
    user_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    pass

class User(UserBase):
    id: int
    profile: Optional[Profile] = None

    class Config:
        orm_mode = True

# API endpoints
@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(username=user.username, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.post("/users/{user_id}/profile/", response_model=Profile)
def create_profile(user_id: int, profile: ProfileCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_profile = Profile(**profile.dict(), user_id=user_id)
    db.add(db_profile)
    db.commit()
    db.refresh(db_profile)
    return db_profile

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

Now you can create a user, add a profile, and then retrieve the user with their associated profile.

One-to-Many Relationship​

A one-to-many relationship is where one record in a table can be associated with multiple records in another table. Let's implement a relationship between User and Post:

python
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    
    # Relationships
    profile = relationship("Profile", back_populates="user", uselist=False)
    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(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    # Relationship - Many-to-One (reverse of One-to-Many)
    author = relationship("User", back_populates="posts")

In this setup:

  • A User can have many Post items (one-to-many)
  • Each Post belongs to one User (many-to-one from the Post perspective)
  • The relationship is bi-directional, meaning we can access posts from a user and the author from a post

Let's add endpoints for creating and retrieving posts:

python
class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    author_id: int

    class Config:
        orm_mode = True

# Update User model to include posts
class User(UserBase):
    id: int
    profile: Optional[Profile] = None
    posts: list[Post] = []

    class Config:
        orm_mode = True

@app.post("/users/{user_id}/posts/", response_model=Post)
def create_post(user_id: int, post: PostCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_post = Post(**post.dict(), author_id=user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

@app.get("/users/{user_id}/posts/", response_model=list[Post])
def get_user_posts(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user.posts

Many-to-Many Relationship​

A many-to-many relationship is where multiple records in one table can be associated with multiple records in another table. For this type of relationship, we need an association table.

Let's implement a relationship between Post and Tag:

python
# Association table
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True)
)

class Tag(Base):
    __tablename__ = "tags"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    
    # Relationship - Many-to-Many
    posts = relationship("Post", secondary=post_tags, back_populates="tags")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    # Relationships
    author = relationship("User", back_populates="posts")
    tags = relationship("Tag", secondary=post_tags, back_populates="posts")

Note the key components:

  • An association table (post_tags) that contains foreign keys to both tables
  • The secondary parameter in the relationship that references the association table
  • Both models have relationships that point back to each other

Let's add endpoints for working with tags:

python
class TagBase(BaseModel):
    name: str

class TagCreate(TagBase):
    pass

class Tag(TagBase):
    id: int
    
    class Config:
        orm_mode = True

# Update Post model to include tags
class Post(PostBase):
    id: int
    author_id: int
    tags: list[Tag] = []

    class Config:
        orm_mode = True

@app.post("/tags/", response_model=Tag)
def create_tag(tag: TagCreate, db: Session = Depends(get_db)):
    db_tag = Tag(**tag.dict())
    db.add(db_tag)
    db.commit()
    db.refresh(db_tag)
    return db_tag

@app.post("/posts/{post_id}/tags/{tag_id}")
def add_tag_to_post(post_id: int, tag_id: int, db: Session = Depends(get_db)):
    post = db.query(Post).filter(Post.id == post_id).first()
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
        
    tag = db.query(Tag).filter(Tag.id == tag_id).first()
    if not tag:
        raise HTTPException(status_code=404, detail="Tag not found")
    
    post.tags.append(tag)
    db.commit()
    return {"message": "Tag added to post"}

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

Advanced Relationship Topics​

Lazy Loading vs. Eager Loading​

By default, SQLAlchemy uses lazy loading, which means related objects are loaded only when accessed. This can lead to the "N+1 query problem" when you access related objects in loops.

To optimize queries, you can use eager loading:

python
# Example of eager loading with joinedload
@app.get("/users/efficient/", response_model=list[User])
def get_users_efficient(db: Session = Depends(get_db)):
    users = db.query(User).options(
        joinedload(User.profile),  # Load profiles eagerly
        joinedload(User.posts)     # Load posts eagerly
    ).all()
    return users

Cascade Delete​

If you want related records to be deleted when the parent is deleted, use the cascade parameter:

python
# Example of cascade delete
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    
    # Delete posts when user is deleted
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")

Self-Referential Relationships​

Sometimes you need to model hierarchical data or relationships within the same table:

python
class Employee(Base):
    __tablename__ = "employees"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    manager_id = Column(Integer, ForeignKey("employees.id"), nullable=True)
    
    # Self-referential relationship
    manager = relationship("Employee", remote_side=[id], backref="subordinates")

Real-World Example: Blog Application​

Let's combine everything we've learned to create a simplified blog API:

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

@app.get("/blog-stats/")
def get_blog_stats(db: Session = Depends(get_db)):
    user_count = db.query(User).count()
    post_count = db.query(Post).count()
    tag_count = db.query(Tag).count()
    
    # Find most used tag
    most_used_tag = db.query(
        Tag.name, 
        func.count(post_tags.c.post_id).label("post_count")
    ).join(post_tags).group_by(Tag.id).order_by(desc("post_count")).first()
    
    # Find most prolific author
    most_prolific_author = db.query(
        User.username, 
        func.count(Post.id).label("post_count")
    ).join(Post).group_by(User.id).order_by(desc("post_count")).first()
    
    return {
        "user_count": user_count,
        "post_count": post_count,
        "tag_count": tag_count,
        "most_used_tag": most_used_tag.name if most_used_tag else None,
        "most_prolific_author": most_prolific_author.username if most_prolific_author else None
    }

@app.get("/search/", response_model=list[Post])
def search_posts(q: str, db: Session = Depends(get_db)):
    return db.query(Post).filter(
        or_(
            Post.title.ilike(f"%{q}%"),
            Post.content.ilike(f"%{q}%")
        )
    ).all()

@app.get("/tags/{tag_name}/posts/", response_model=list[Post])
def get_posts_by_tag(tag_name: str, db: Session = Depends(get_db)):
    tag = db.query(Tag).filter(Tag.name == tag_name).first()
    if not tag:
        raise HTTPException(status_code=404, detail="Tag not found")
    return tag.posts

This example demonstrates how you can use relationships to build powerful queries and features in your API.

Summary​

In this tutorial, we've covered:

  1. The three main types of database relationships: one-to-one, one-to-many, and many-to-many
  2. How to implement each type with SQLAlchemy ORM in FastAPI
  3. How to create Pydantic models that work with these relationships
  4. Advanced topics like eager loading and cascade deletion
  5. A real-world example bringing everything together

Understanding how to properly model and work with relationships is crucial for building scalable and maintainable FastAPI applications.

Exercises​

To reinforce your knowledge, try these exercises:

  1. Create a comment system for the blog (where users can comment on posts)
  2. Implement a "like" system for posts (using a many-to-many relationship between users and posts)
  3. Add categories to posts (where each post belongs to one category, but categories can have many posts)
  4. Create an API endpoint that returns a user's feed (posts from authors they follow)

Additional Resources​

By mastering ORM relationships in FastAPI, you'll be able to model complex data interactions and build more powerful applications.


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