Dive Into CRUD With FastAPI, SQLAlchemy, and Pydantic

Creating CRUD (Create, Read, Update, Delete) endpoints is like the bread and butter of building web apps. If you’re using FastAPI along with SQLAlchemy and Pydantic, you’re setting yourself up for a smooth ride. Let’s break down how to whip up a CRUD app with these tools, and don’t worry — we’ll keep things chill and easy to follow.

Getting Your Playground Ready

First things first, you need to make sure you’ve got all the necessary goodies installed. Use pip for that:

pip install fastapi uvicorn sqlalchemy psycopg2-binary

In this example, we’re rolling with SQLite because it’s straightforward, but if you vibe more with PostgreSQL or something else, switch it up however you like.

Building the Database Models

Alright, let’s start by creating our database models using SQLAlchemy. Think of these models as blueprints for your database tables.

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

Crafting Pydantic Models

Next up, we gotta define some Pydantic models. These schemas help in validating and serializing the data. They should mirror your database models but are used specifically for API interactions.

from pydantic import BaseModel

class ItemBase(BaseModel):
    name: str
    description: str

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int

    class Config:
        orm_mode = True

Setting Up FastAPI With Style

Now it’s time to roll up our sleeves and set up FastAPI and define those CRUD endpoints.

from fastapi import FastAPI, Depends
from fastapi.responses import JSONResponse
from fastapi.requests import Request
from fastapi.exceptions import HTTPException

app = FastAPI(lifespan=lifespan)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

@app.post("/items/", response_model=Item)
async def create_item(item: ItemCreate, session: AsyncSession = Depends(get_session)):
    db_item = Item(name=item.name, description=item.description)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

@app.get("/items/", response_model=list[Item])
async def read_items(session: AsyncSession = Depends(get_session)):
    items = await session.execute(select(Item))
    return items.scalars().all()

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
    item = await session.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: ItemCreate, session: AsyncSession = Depends(get_session)):
    db_item = await session.get(Item, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    db_item.name = item.name
    db_item.description = item.description
    await session.commit()
    await session.refresh(db_item)
    return db_item

@app.delete("/items/{item_id}")
async def delete_item(item_id: int, session: AsyncSession = Depends(get_session)):
    item = await session.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    await session.delete(item)
    await session.commit()
    return JSONResponse(content={"message": "Item deleted"}, status_code=200)

Let’s Get This Show on the Road

To get everything up and running, just use this command:

uvicorn main:app --reload

Your FastAPI server will come alive, and you can start playing with your CRUD endpoints at http://localhost:8000.

Automated Swagger Documentation

One of the coolest things about FastAPI is the automated documentation it provides through Swagger UI. You can check it out at http://localhost:8000/docs and test your API endpoints right from your browser.

Taking It Up a Notch with FastCRUD

If you’re feeling adventurous and want even more streamlined CRUD operations, try using packages like FastCRUD. It brings features like automatic endpoint creation, dynamic query building, and pagination to the table, simplifying complex CRUD operations.

from fastcrud import FastCRUD, crud_router

item_router = crud_router(
    session=get_session,
    model=Item,
    create_schema=ItemCreate,
    update_schema=ItemUpdateSchema,
    path="/items",
    tags=["Items"],
)

app.include_router(item_router)

With this setup, you get CRUD endpoints for your Item model on autopilot, including support for joins, sorting, and pagination.

Wrapping Up

Building CRUD endpoints with FastAPI, SQLAlchemy, and Pydantic is super effective for managing data in your web apps. By following the steps above, you’ll have a robust and efficient API handling your basic operations — creating, reading, updating, and deleting data. Whether you stick with the basics or dive into advanced tools like FastCRUD, FastAPI’s got your back with a flexible and high-performance framework to meet all your needs. Happy coding!