Prototyping FastAPI faster with FastCRUD

You probably already know about FastAPI, but in case you don't - it's a modern python web framework built on top of Starlette and Pydantic with cool features like Async support and auto generated swagger documentation.

Creating a minimal FastAPI app is as simple as:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

FastAPI gained a lot of popularity by being fast compared to other python web frameworks, having modern features, and great documentation, but it is not a batteries included framework like Django so you need to use other libraries for stuff like database operations.

Let's create a more functional FastAPI application with database interaction using SQLModel. We start by actually creating a connection to the database:

FastAPI Basic Setup

# database.py

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)

async_session = sessionmaker(
  engine, 
  class_=AsyncSession, 
  expire_on_commit=False
)

And creating a function to actually get this database session

# database.py
...
async def async_get_db() -> AsyncSession:
    async_session = local_session

    async with async_session() as db:
        yield db
        await db.commit()

Now we can finally create our database models. I'll keep it simple, a user can create posts and that's it. Posts have a title and the actual text.

# models.py

from sqlmodel import Field, SQLModel, Relationship
from typing import Optional, List
from datetime import datetime

class User(SQLModel, table=True):
    __tablename__ = "user"

    id: Optional[int] = Field(default=None, primary_key=True, nullable=False, index=True)
    name: str = Field(max_length=30)
    username: str = Field(max_length=20, index=True, unique=True)

    posts: List["Post"] = Relationship(back_populates="creator")

class Post(SQLModel, table=True):
    __tablename__ = "post"

    id: Optional[int] = Field(default=None, primary_key=True, nullable=False, index=True)
    created_by_user_id: int = Field(default=None, foreign_key="user.id")
    title: str = Field(max_length=30)
    text: str = Field(max_length=63206)

    creator: User = Relationship(back_populates="posts")

Now let's create the schemas that we'll use to validate our input and output:

# user_schemas.py

from typing import List, Optional
from sqlmodel import Field, SQLModel

# Shared properties
class UserBase(SQLModel):
    name: Optional[str] = None
    username: Optional[str] = None

# Properties to receive via API on creation
class UserCreate(UserBase):
    pass  

# Properties to receive via API on update
class UserUpdate(UserBase):
    pass

# Properties to return to client
class UserRead(UserBase):
    id: Optional[int] = None
    posts: List["PostRead"] = []

You should similarly create the schemas for posts, but I'm skipping it since it's similar.

Now we'll finally create our FastAPI app:

# main.py

from .database.py import SQLModel
from fastapi import FastAPI

# We want to create the tables on start, so let's create this function
async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

app = FastAPI()

# now adding the event handler:
app.add_event_handler("startup", create_tables)

Creating Endpoints

We finally have our basic setup done, now let's create the user endpoints for our app, you'll write similar code for each model you add to the api.

# user_endpoints.py

import fastapi
from fastapi import HTTPException, Depends, status
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from typing import List, Optional

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

router = fastapi.APIRouter(tags=["Users"])

@app.post("/users/", response_model=UserRead)
async def create_user(user: UserCreate, session: AsyncSession = Depends(get_session)):
    db_user = User.from_orm(user)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

@app.get("/users/", response_model=List[UserRead])
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

@app.get("/users/{user_id}", response_model=UserRead)
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.patch("/users/{user_id}", response_model=UserRead)
async def update_user(user_id: int, user: UserUpdate, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    db_user = result.scalar_one_or_none()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    user_data = user.dict(exclude_unset=True)
    for key, value in user_data.items():
        setattr(db_user, key, value)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

This works, but a lot of the time you are just writing the same code again and again. You have the session transactions, filters, updating fields, committing changes to the database…

Creating Endpoints with FastCRUD

Enter FastCRUD - a package for FastAPI, offering robust async CRUD operations and flexible endpoint creation utilities.
The same code above using FastCRUD would be written as:

# user_endpoints.py

import fastapi
from fastcrud import FastCRUD
from fastcrud.exceptions.http_exceptions import (
    DuplicateValueException, 
    NotFoundException
)

router = fastapi.APIRouter(tags=["Users"])

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

crud_user = FastCRUD(User)

@app.post("/users/", response_model=UserRead)
async def create_user(
    user: UserCreate, 
    db: AsyncSession = Depends(get_session)
):
    username_row = await crud_user.exists(db=db, username=user.username)
    if username_row:
        raise DuplicateValueException("Username not available")
    return await crud_user.create(db=db, object=user)

@app.get("/users/", response_model=List[UserRead])
async def read_users(
    db: AsyncSession = Depends(get_session), 
    page: int = 1, 
    items_per_page: int = 10
):
    return await crud_user.get_multi(
        db=db,
        offset=((page - 1) * items_per_page),
        limit=items_per_page,
        schema_to_select=UserRead
    )

@app.get("/users/{user_id}", response_model=UserRead)
async def read_user(
    user_id: int, 
    db: AsyncSession = Depends(get_session)
):
    db_user = await crud_user.get(
        db=db, 
        schema_to_select=UserRead, 
        id=user_id
    )
    if db_user is None:
        raise NotFoundException("User not found")

    return db_user

@app.patch("/users/{user_id}", response_model=UserRead)
async def update_user(
    user_id: int, 
    user: UserUpdate, 
    db: AsyncSession = Depends(get_session)
    ):
    db_user = await crud_users.get(
        db=db, 
        schema_to_select=UserRead, 
        username=username
    )
    if db_user is None:
        raise NotFoundException("User not found")

    return await crud_user.update(db=db, object=user, id=user_id)

When you instantiate FastCRUD with a Model, you automatically get a bunch of methods for robust database interaction. A complete list of the methods and how they work here, but the most common are:

• Create
• Get
• Exists
• Count
• Get Multi
• Update

Automatic Endpoints

Sometimes, even this is not quick enough. When you are just prototyping or at a hackathon, you really want to do this as fast as you can. That's why FastCRUD provides crud_router, a way to instantly have your endpoints ready for prototyping:

# user_endpoints.py

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

from fastcrud import FastCRUD, crud_router

# CRUD router setup
user_router = crud_router(
    session=async_session,
    model=User,
    crud=FastCRUD(User),
    create_schema=UserCreate,
    update_schema=UserUpdate,
    tags=["Users"]
)

Let's add this router to our application:

# main.py

# now also importing our user_router
from .user_endpoints import user_router
from .database.py import SQLModel

from fastapi import FastAPI

async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

app = FastAPI()
app.add_event_handler("startup", create_tables)

# including the user_router
app.include_router(user_router)

And that's it. As simple as that, you may go to /docs and you'll see the created CRUD endpoints. If you have a more custom usage, just check the docs.

Full disclaimer: I'm the creator and maintainer of FastCRUD, I am sharing it because it's a solution for a problem of mine that I think a lot of people could benefit from.

FAQ

Q1: Where Can I read more about FastCRUD?
Just go to the docs: https://igorbenav.github.io/fastcrud/

Q2: Supported ORMs
Currently, FastCRUD works with SQLAlchemy 2.0 or greater and SQLModel 0.14 or greater.

Q3: Where can I find support or contribute to FastCRUD?
If you have any question or issue, just head to the Github repo and open an issue.

Connect With Me

If you have any questions, want to discuss tech-related stuff or simply share your feedback, feel free to reach me on social media

Github: igorbenav
X: igorbenav
Linkedin: Igor