Python Fastapi Async Crud Operations Tutorial

Welcome to the ultimate Python FastAPI async CRUD operations tutorial! In this guide you’ll learn how to build a high‑performance REST API that handles Create, Read, Update, and Delete (CRUD) actions using FastAPI’s native async support. Whether you’re a seasoned Python developer or just starting out, this step‑by‑step walkthrough will equip you with everything you need to design, code, and test a production‑ready async API in under an hour.

Why Choose FastAPI for Async CRUD?

FastAPI has quickly become the go‑to framework for modern web services because it combines speed, developer friendliness, and automatic OpenAPI documentation. Its async capabilities let you handle thousands of concurrent requests without blocking the event loop, making it perfect for I/O‑bound tasks like database interactions.

• Performance: Comparable to Node.js and Go, thanks to Starlette and Pydantic.
• Type safety: Python type hints are validated at runtime, reducing bugs.
• Built‑in docs: Swagger UI and ReDoc appear automatically.
• Scalability: Async functions let you scale horizontally with minimal code changes.

Prerequisites and Environment Setup

1. Install Python 3.9+ and create a virtual environment

python3 -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`

2. Install FastAPI, Uvicorn, and an async‑compatible ORM

For this tutorial we’ll use SQLModel, which blends Pydantic and SQLAlchemy and works natively with async engines.

pip install fastapi uvicorn[standard] sqlmodel aiosqlite

3. Project structure

• app/main.py – FastAPI application entry point.
• app/models.py – Database models.
• app/crud.py – Async CRUD helper functions.
• app/router.py – API routes.

Defining the Data Model with SQLModel

SQLModel lets you declare tables using Pydantic‑style classes. Below is a simple Item model representing a product in an inventory.

# app/models.py
from typing import Optional
from sqlmodel import SQLModel, Field

class Item(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True, max_length=100)
    description: Optional[str] = Field(default=None, max_length=255)
    price: float
    in_stock: bool = Field(default=True)

Creating an Async Database Engine

FastAPI works best with an async engine. We’ll use SQLite for simplicity, but the same pattern applies to PostgreSQL, MySQL, or any async‑compatible DB.

# app/database.py
from sqlmodel import create_engine, SQLModel
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

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

engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=True)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

Building Async CRUD Functions

Keeping business logic separate from route definitions makes the code easier to test and maintain. Each function below uses async with to acquire a database session without blocking the event loop.

Create an item

# app/crud.py
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from .models import Item

async def create_item(session: AsyncSession, item: Item) -> Item:
    session.add(item)
    await session.commit()
    await session.refresh(item)
    return item

Read (list & single)

async def get_item(session: AsyncSession, item_id: int) -> Item | None:
    result = await session.get(Item, item_id)
    return result

async def list_items(session: AsyncSession, skip: int = 0, limit: int = 100) -> list[Item]:
    statement = select(Item).offset(skip).limit(limit)
    results = await session.exec(statement)
    return results.all()

Update an item

async def update_item(session: AsyncSession, item_id: int, data: dict) -> Item | None:
    db_item = await session.get(Item, item_id)
    if not db_item:
        return None
    for key, value in data.items():
        setattr(db_item, key, value)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

Delete an item

async def delete_item(session: AsyncSession, item_id: int) -> bool:
    db_item = await session.get(Item, item_id)
    if not db_item:
        return False
    await session.delete(db_item)
    await session.commit()
    return True

Exposing the CRUD Operations via FastAPI Routes

Now we tie everything together in a router that FastAPI will serve. Notice the use of Depends to inject an async session into each endpoint.

# app/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel.ext.asyncio.session import AsyncSession
from .database import engine
from .models import Item
from .crud import create_item, get_item, list_items, update_item, delete_item

router = APIRouter(prefix="/items", tags=["items"])

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def api_create_item(item: Item, session: AsyncSession = Depends(get_session)):
    return await create_item(session, item)

@router.get("/", response_model=list[Item])
async def api_list_items(skip: int = 0, limit: int = 100,
                         session: AsyncSession = Depends(get_session)):
    return await list_items(session, skip, limit)

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

@router.put("/{item_id}", response_model=Item)
async def api_update_item(item_id: int, item: Item,
                          session: AsyncSession = Depends(get_session)):
    updated = await update_item(session, item_id, item.dict(exclude_unset=True))
    if not updated:
        raise HTTPException(status_code=404, detail="Item not found")
    return updated

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def api_delete_item(item_id: int, session: AsyncSession = Depends(get_session)):
    success = await delete_item(session, item_id)
    if not success:
        raise HTTPException(status_code=404, detail="Item not found")
    return None

Launching the Application

Finally, create the FastAPI app, include the router, and start the server with Uvicorn.

# app/main.py
import uvicorn
from fastapi import FastAPI
from .router import router
from .database import init_db

app = FastAPI(
    title="Async CRUD API with FastAPI",
    description="A minimal example demonstrating async CRUD operations using FastAPI and SQLModel.",
    version="1.0.0",
)

app.include_router(router)

@app.on_event("startup")
async def on_startup():
    await init_db()

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

Testing Your Async Endpoints

With the server running (uvicorn app.main:app --reload), you can quickly verify each route using curl or httpie. Below are example commands:

1. Create a new item:

   http POST http://localhost:8000/items/ name="Widget" price:=19.99 description="A useful widget"

2. List all items:

   http GET http://localhost:8000/items/

3. Retrieve a single item (replace 1 with the actual ID):

   http GET http://localhost:8000/items/1

4. Update an item:
   <