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 defendpoints 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_sizeandmax_overflowin 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