Prototyping FastAPI faster with FastCRUD
Tags: fastapi, python, web, backend
Published: 2024-01-31 16:18:56

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:

```python
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

```python
# 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

```python
# 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.

```python
# 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:

```python
# 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:

```python
# 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.

```python
# 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:

```python
# 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](https://igorbenav.github.io/fastcrud/usage/crud/), 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:

```python
# 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:

```python
# 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](https://github.com/igorbenav/fastcrud) 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](github.com/igorbenav)
X: [igorbenav](x.com/igorbenav)
Linkedin: [Igor](https://www.linkedin.com/in/igor-magalhaes-r/)