Introduction to SQLAlchemy in FastAPI

When building robust web applications with FastAPI, integrating a database is often a crucial requirement. SQLAlchemy, a powerful and flexible Object-Relational Mapping (ORM) tool, provides an excellent solution for database integration in Python. In this blog post, we'll explore how to use SQLAlchemy with FastAPI to create efficient and scalable database-driven applications.

Setting Up SQLAlchemy with FastAPI

To get started, you'll need to install the necessary dependencies:

pip install fastapi sqlalchemy databases

Next, let's set up the database connection and create a base model:

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

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Defining Models

With the database connection set up, we can now define our models. Let's create a simple User model:

from sqlalchemy import Column, Integer, String

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)
    hashed_password = Column(String)

Creating Database Tables

To create the database tables based on our models, we need to call Base.metadata.create_all(engine) when starting our application:

def create_tables():
    Base.metadata.create_all(bind=engine)

if __name__ == "__main__":
    create_tables()
    uvicorn.run(app, host="0.0.0.0", port=8000)

Dependency Injection for Database Sessions

To ensure proper handling of database sessions, we'll create a dependency that yields a database session:

from fastapi import Depends
from sqlalchemy.orm import Session

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

CRUD Operations

Now that we have our database set up, let's implement some CRUD (Create, Read, Update, Delete) operations:

Create

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas

router = APIRouter()

@router.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = models.User(username=user.username, email=user.email, hashed_password=user.password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Read

@router.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

Update

@router.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, 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")
    for key, value in user.dict().items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user

Delete

@router.delete("/users/{user_id}", response_model=schemas.User)
def delete_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")
    db.delete(db_user)
    db.commit()
    return db_user

Handling Relationships

SQLAlchemy makes it easy to define and work with relationships between models. Let's add a Post model and establish a one-to-many relationship with the User model:

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

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)
    hashed_password = Column(String)

    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"))

    author = relationship("User", back_populates="posts")

Now you can easily create posts for users and retrieve all posts for a specific user:

@router.post("/users/{user_id}/posts/", response_model=schemas.Post)
def create_post_for_user(user_id: int, post: schemas.PostCreate, 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_post = models.Post(**post.dict(), author_id=user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

@router.get("/users/{user_id}/posts/", response_model=List[schemas.Post])
def read_user_posts(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.posts

Advanced Querying

SQLAlchemy provides powerful querying capabilities. Here are some examples of more advanced queries:

Filtering and Ordering

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

Joining Tables

@router.get("/users-with-posts/", response_model=List[schemas.UserWithPosts])
def read_users_with_posts(db: Session = Depends(get_db)):
    return db.query(models.User).options(joinedload(models.User.posts)).all()

Aggregations

from sqlalchemy import func

@router.get("/user-post-counts/", response_model=List[schemas.UserPostCount])
def read_user_post_counts(db: Session = Depends(get_db)):
    return db.query(models.User.id, models.User.username, func.count(models.Post.id).label("post_count")).\
        join(models.Post).\
        group_by(models.User.id).\
        all()

Conclusion

By integrating SQLAlchemy with FastAPI, you can create powerful, database-driven applications with ease. This combination allows for efficient data management, complex querying, and seamless integration with your FastAPI routes. As you continue to explore SQLAlchemy, you'll discover even more advanced features that can help you build robust and scalable web applications.