Python Fastapi Async Crud Operations Tutorial

Welcome to the ultimate Python FastAPI async CRUD operations tutorial! If you’re looking to build high‑performance APIs that can handle thousands of concurrent requests, FastAPI is the perfect framework. In this guide, we’ll walk through everything you need to know—from setting up an async‑ready project to implementing Create, Read, Update, and Delete (CRUD) endpoints with proper validation, database integration, and best‑practice patterns. By the end, you’ll have a production‑ready codebase you can extend for any web service.

Why Choose FastAPI for Async CRUD?

FastAPI combines several modern Python features that make async CRUD development a breeze:

• Native async support: Built on Starlette, FastAPI lets you write async def endpoints that run on an event loop, delivering non‑blocking I/O.
• Automatic data validation: Pydantic models validate request bodies and query parameters out of the box.
• Interactive documentation: Swagger UI and ReDoc are generated automatically, saving you time on API docs.
• Speed: Benchmarks show FastAPI rivals Node.js and Go for throughput, especially when using async database drivers.

Project Setup – Getting Started in Minutes

1. Create a virtual environment

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

2. Install required packages

pip install fastapi[all] uvicorn sqlalchemy aiosqlite asyncpg

We’ll use SQLAlchemy 2.0 with the async extension, aiosqlite for a quick SQLite demo, and asyncpg if you switch to PostgreSQL later.

3. Project structure

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── schemas.py
│   └── crud.py
└── requirements.txt

Database Configuration – Async SQLAlchemy

First, define the async engine and session maker in db.py:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test.db"  # Change to PostgreSQL if needed

engine = create_async_engine(DATABASE_URL, echo=True)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Base = declarative_base()

Defining Models and Schemas

SQLAlchemy model

# models.py
from sqlalchemy import Column, Integer, String
from .db import Base

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    description = Column(String, nullable=True)
    price = Column(Integer, nullable=False)

Pydantic schemas

# schemas.py
from pydantic import BaseModel, Field

class ItemBase(BaseModel):
    name: str = Field(..., example="Laptop")
    description: str | None = Field(None, example="A powerful gaming laptop")
    price: int = Field(..., gt=0, example=1500)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    price: int | None = None

class ItemRead(ItemBase):
    id: int

    class Config:
        orm_mode = True

CRUD Utility Functions

Keeping database logic separate from the API layer improves testability. In crud.py we implement async CRUD helpers:

# crud.py
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas

async def get_item(db: AsyncSession, item_id: int):
    result = await db.execute(select(models.Item).where(models.Item.id == item_id))
    return result.scalar_one_or_none()

async def get_items(db: AsyncSession, skip: int = 0, limit: int = 100):
    result = await db.execute(select(models.Item).offset(skip).limit(limit))
    return result.scalars().all()

async def create_item(db: AsyncSession, item: schemas.ItemCreate):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

async def update_item(db: AsyncSession, item_id: int, item: schemas.ItemUpdate):
    stmt = (
        update(models.Item)
        .where(models.Item.id == item_id)
        .values(**{k: v for k, v in item.dict().items() if v is not None})
        .execution_options(synchronize_session="fetch")
    )
    await db.execute(stmt)
    await db.commit()
    return await get_item(db, item_id)

async def delete_item(db: AsyncSession, item_id: int):
    stmt = delete(models.Item).where(models.Item.id == item_id)
    await db.execute(stmt)
    await db.commit()
    return {"ok": True}

FastAPI Endpoints – Wiring Everything Together

Now we can expose the CRUD operations via async routes in main.py:

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from . import db, models, schemas, crud

app = FastAPI(title="Async CRUD with FastAPI", version="1.0.0")

# Dependency that provides an async DB session
async def get_db() -> AsyncSession:
    async with db.AsyncSessionLocal() as session:
        yield session

@app.on_event("startup")
async def startup():
    # Create tables on startup (only for demo; use Alembic in prod)
    async with db.engine.begin() as conn:
        await conn.run_sync(models.Base.metadata.create_all)

@app.post("/items/", response_model=schemas.ItemRead, status_code=status.HTTP_201_CREATED)
async def create_item(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    return await crud.create_item(db, item)

@app.get("/items/", response_model=list[schemas.ItemRead])
async def read_items(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    return await crud.get_items(db, skip, limit)

@app.get("/items/{item_id}", response_model=schemas.ItemRead)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    db_item = await crud.get_item(db, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

@app.put("/items/{item_id}", response_model=schemas.ItemRead)
async def update_item(item_id: int, item: schemas.ItemUpdate, db: AsyncSession = Depends(get_db)):
    db_item = await crud.get_item(db, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return await crud.update_item(db, item_id, item)

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, db: AsyncSession = Depends(get_db)):
    db_item = await crud.get_item(db, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    await crud.delete_item(db, item_id)
    return None

Running the Application

Start the server with Uvicorn, which is an ASGI‑compatible web server optimized for async workloads:

uvicorn app.main:app --reload

The --reload flag watches for file changes, making development fast and painless. Open http://127.0.0.1:8000/docs to explore the automatically generated Swagger UI, or /redoc for ReDoc.

Testing Async Endpoints

FastAPI works seamlessly with pytest and httpx for async testing. Below is a minimal test suite:

# test_main.py
import pytest
from httpx import AsyncClient
from app.main import app
from app.db import engine, Base

@pytest.fixture(scope="module", autouse=True)
async def create_test_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.mark.asyncio
async def test_create_item():
    async with AsyncClient(app=app, base_url="http://test") as client:
        payload = {"name": "Keyboard", "price": 99}
        response = await client.post("/items/", json=payload)
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Keyboard"
        assert data["price"] == 99

Best Practices for Production‑Ready Async CRUD

• Use a real async driver: Switch from SQLite/aiosqlite to PostgreSQL/asyncpg for scalability.
• Employ Alembic migrations: Manage schema changes without dropping tables on startup.
• Configure connection pooling: Adjust pool_size and max_overflow in the engine URL for optimal throughput.
• Validate input thoroughly: Leverage Pydantic validators for complex business rules.
• Secure endpoints: Add OAuth2, JWT, or API key dependencies using FastAPI’s security utilities.
• Implement logging and monitoring: Integrate structured logs (e.g., Loguru