Welcome to our Python FastAPI async CRUD operations tutorial! If you’re looking to build high‑performance, non‑blocking web APIs with Python, FastAPI is the perfect framework. In this guide, we’ll walk you through setting up a FastAPI project, connecting to an asynchronous database, and implementing full Create, Read, Update, Delete (CRUD) functionality using async/await. By the end, you’ll have a production‑ready codebase you can extend for any modern web service.
Why Choose FastAPI for Async CRUD?
- Speed: Built on Starlette and Pydantic, FastAPI delivers near‑native performance, often rivaling Node.js and Go.
- Async‑first: Native support for
asyncendpoints means you can handle thousands of concurrent requests without blocking. - Automatic docs: OpenAPI and Swagger UI are generated automatically, making testing and client integration effortless.
- Type safety: Python type hints improve code readability and catch bugs early, especially when dealing with complex data models.
Project Setup: Tools and Dependencies
Before diving into code, make sure you have the following installed:
python>=3.9
pip
virtualenv (optional but recommended)Next, create a virtual environment and install the required packages:
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install fastapi[all] uvicorn sqlalchemy[asyncio] aiosqliteWe’ll use SQLAlchemy with its async extension and aiosqlite as the SQLite driver for simplicity. In a real‑world scenario, you could swap SQLite for PostgreSQL (via asyncpg) without changing much code.
Defining the Database Model
Let’s create a simple Item model that represents a product in an inventory system.
# models.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy import Column, Integer, String, Text
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
bind=engine, class_=AsyncSession, expire_on_commit=False
)
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), unique=True, index=True, nullable=False)
description = Column(Text, nullable=True)
price = Column(Integer, nullable=False)
Creating the Database Tables
Run the following script once to create the tables:
# create_db.py
import asyncio
from models import Base, engine
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
if __name__ == "__main__":
asyncio.run(init_db())Execute it with python create_db.py. You’ll see SQL statements printed to the console, confirming that the items table is ready.
Building the FastAPI Application
Now we’ll set up the FastAPI app, define Pydantic schemas for request/response validation, and implement the async CRUD endpoints.
Pydantic Schemas
# schemas.py
from pydantic import BaseModel, Field
class ItemBase(BaseModel):
name: str = Field(..., max_length=100, description="Unique name of the item")
description: str | None = Field(None, description="Optional detailed description")
price: int = Field(..., gt=0, description="Price in cents")
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
Dependency for Database Session
# deps.py
from typing import AsyncGenerator
from models import AsyncSessionLocal
async def get_db() -> AsyncGenerator:
async with AsyncSessionLocal() as session:
yield session
CRUD Operations
Below is the core logic for each CRUD operation. Notice the use of await with SQLAlchemy’s async methods.
# crud.py
from sqlalchemy import select, update, delete
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from models import Item
from schemas import ItemCreate, ItemUpdate
async def get_item(db: AsyncSession, item_id: int):
result = await db.execute(select(Item).where(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(Item).offset(skip).limit(limit))
return result.scalars().all()
async def create_item(db: AsyncSession, item: ItemCreate):
db_item = Item(**item.dict())
db.add(db_item)
try:
await db.commit()
await db.refresh(db_item)
return db_item
except IntegrityError:
await db.rollback()
raise
async def update_item(db: AsyncSession, item_id: int, item: ItemUpdate):
stmt = (
update(Item)
.where(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):
await db.execute(delete(Item).where(Item.id == item_id))
await db.commit()
return {"detail": f"Item {item_id} deleted"}
Routing: Async Endpoints in FastAPI
Finally, wire everything together in the main application file.
# main.py
from fastapi import FastAPI, HTTPException, Depends, status
from typing import List
from models import Base, engine
from schemas import ItemCreate, ItemRead, ItemUpdate
from crud import (
get_item,
get_items,
create_item,
update_item,
delete_item,
)
from deps import get_db
app = FastAPI(
title="FastAPI Async CRUD Tutorial",
description="A step‑by‑step guide to building async CRUD APIs with FastAPI and SQLAlchemy.",
version="1.0.0",
)
@app.on_event("startup")
async def startup():
# Ensure tables exist on startup (optional)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@app.get("/items/", response_model=List[ItemRead])
async def read_items(skip: int = 0, limit: int = 10, db=Depends(get_db)):
items = await get_items(db, skip=skip, limit=limit)
return items
@app.get("/items/{item_id}", response_model=ItemRead)
async def read_item(item_id: int, db=Depends(get_db)):
item = await get_item(db, item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
@app.post("/items/", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
async def create_new_item(item: ItemCreate, db=Depends(get_db)):
try:
new_item = await create_item(db, item)
return new_item
except Exception:
raise HTTPException(
status_code=400,
detail="Item could not be created – possibly a duplicate name.",
)
@app.put("/items/{item_id}", response_model=ItemRead)
async def update_existing_item(
item_id: int, item: ItemUpdate, db=Depends(get_db)
):
existing = await get_item(db, item_id)
if not existing:
raise HTTPException(status_code=404, detail="Item not found")
updated = await update_item(db, item_id, item)
return updated
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_existing_item(item_id: int, db=Depends(get_db)):
existing = await get_item(db, item_id)
if not existing:
raise HTTPException(status_code=404, detail="Item not found")
await delete_item(db, item_id)
return None
Run the server with:
uvicorn main:app --reloadVisit http://127.0.0.1:8000/docs to explore the automatically generated Swagger UI. You can test each CRUD endpoint directly from the browser.
Best Practices for Production‑Ready Async FastAPI Apps
- Connection pooling: When using PostgreSQL, configure
asyncpgpool size to match your workload. - Input validation: Leverage Pydantic’s validators for complex fields (e.g., price formats, email validation).
- Error handling: Centralize exception handling with FastAPI’s
exception_handlerdecorators. - Security: Add OAuth2, JWT, or API key authentication to protect your CRUD routes.
- Testing: Use
httpx.AsyncClientandpytest-asynciofor async endpoint tests. - Logging: Integrate structured logging (e.g.,
structlog) to capture request traces.
Scaling Async FastAPI with Docker and Kubernetes
Containerizing your FastAPI service ensures consistency across environments. Below is a minimal Dockerfile for the async CRUD app:
# Dockerfile
FROM python:3.11
Leave a Reply