Retriving from Database in FastAPI

Today we're going to implement a super common pattern - getting a row or model object from database by its slug.

Let's start with the crud layer at database/crud/blog.py

from sqlmodel import Session
from fastapi import Depends
from sqlmodel import select

from database.models.blog import Blog
from apis.deps import get_db
from schemas.blog import CreateBlog


def insert_blog(blog: CreateBlog, db: Session) -> Blog:
    #same same


def get_blog_by_slug(slug: str, db: Session) -> Blog:
    return db.query(Blog).filter(Blog.slug == slug).first()

Let's Break It Down

The Helper Function: get_blog_by_slug() is doing the heavy lifting here. We're using the old-school SQLAlchemy way with .query() and .filter(). Yeah, I know SQLModel has that fancy select() syntax, but sometimes the classic way just feels more readable. I will be demonstrating the SQLModel way too, just wait for a while.

Now, we can concentrate on the route too at apis/v1/blog.py:

@router.get("/blogs/{slug}", response_model=ShowBlog)
def get_blog(slug: str, db: Session = Depends(get_db)):
    return get_blog_by_slug(slug, db)

The Route: Our actual endpoint is super clean. Just grab the slug from the URL path, pass it to our helper function, and boom - FastAPI automatically converts our Blog model to the ShowBlog response format.

I love this approach because:

Separation of Concerns: Database logic stays in the helper function, route logic stays in the route. Clean and organized!
Reusability: Need to get a blog by slug somewhere else? Just import get_blog_by_slug().
Testing: Way easier to test a pure function that takes a slug and returns a blog than testing the entire route.

If your company has already adopted to SQLModel, then you might want to refactor the crud layer to follow the SQLModel way. The only change is SQLModel exposes a different approach of using select() statement and it is a bit more readable to me because of the where function instead of filter.

from sqlmodel import Session
from fastapi import Depends
from sqlmodel import select

from database.models.blog import Blog
from apis.deps import get_db
from schemas.blog import CreateBlog


def insert_blog(blog: CreateBlog, db: Session) -> Blog:
    db_blog = Blog(
        title=blog.title,
        slug=blog.slug,
        content=blog.content,
        is_active=True,
        user_id=1
    )
    db.add(db_blog)
    db.commit()
    db.refresh(db_blog)
    return db_blog


def get_blog_by_slug(slug: str, db: Session) -> Blog:
    # return db.query(Blog).filter(Blog.slug == slug).first()  #the sqlalchemy way
    statement = select(Blog).where(Blog.slug == slug)
    result = db.exec(statement)
    return result.first()

One Thing That Bugs Me

See that .first() at the end? If no blog is found, this returns None. Your frontend is going to get a 200 response with null data, which is... weird. Let's add some exception handling to improve our API.

from fastapi import APIRouter, Depends, status, HTTPException

@router.get("/blogs/{slug}", response_model=ShowBlog)
def get_blog(slug: str, db: Session = Depends(get_db)):
    blog = get_blog_by_slug(slug, db)
    if not blog:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Blog not found")
    return blog

There is still a ton of work that could be done to improve the API, e.g. we can use caching, rate limiting etc. But let's go in an incremental manner and do things in small small chunks.

Retriving from Database in FastAPI | SQLModel

Let's Break It Down

One Thing That Bugs Me