arunakasingh.in

Python Fastapi Async Crud Operations Tutorial

Written by

arun

in

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 framework you need. In this guide we’ll walk through setting up a FastAPI project, connecting to an async database, and implementing full‑featured Create, Read, Update, and Delete (CRUD) endpoints—all using Python’s async/await syntax. By the end, you’ll have a production‑ready codebase you can extend for any web application.

Why Choose FastAPI for Async CRUD?

  • Speed: Built on Starlette and Pydantic, FastAPI delivers near‑C++ performance while staying pure Python.
  • Async‑first: Native support for asyncio lets you write non‑blocking I/O code without extra libraries.
  • Automatic docs: OpenAPI and Swagger UI are generated automatically, making testing a breeze.
  • Type safety: Python type hints are enforced at runtime, reducing bugs early.

Project Setup

1. Install Dependencies

python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`
pip install fastapi[all] uvicorn sqlalchemy[asyncio] aiosqlite

We use uvicorn as the ASGI server, SQLAlchemy with async support for ORM, and aiosqlite as a lightweight SQLite driver (you can swap it for PostgreSQL or MySQL later).

2. Project Structure

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

Defining the Database Model

First, we create a simple Item model that represents a product in an inventory.

# app/models.py
from sqlalchemy import Column, Integer, String, Text
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

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

Creating Pydantic Schemas

FastAPI uses Pydantic models to validate request bodies and shape responses.

# app/schemas.py
from pydantic import BaseModel, Field
from typing import Optional

class ItemBase(BaseModel):
    name: str = Field(..., max_length=100, example="Widget")
    description: Optional[str] = Field(None, example="A useful widget")
    price: int = Field(..., gt=0, example=1999)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, max_length=100)
    description: Optional[str] = None
    price: Optional[int] = Field(None, gt=0)

class ItemOut(ItemBase):
    id: int

    class Config:
        orm_mode = True

Async Database Session

We configure an async engine and a dependency that yields a session for each request.

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

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

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Implementing CRUD Functions

All database interactions are defined in crud.py using async functions.

# app/crud.py
from sqlalchemy.future import select
from sqlalchemy import update, delete
from sqlalchemy.exc import IntegrityError
from .models import Item
from .schemas import ItemCreate, ItemUpdate
from typing import List, Optional

async def get_item(db, item_id: int) -> Optional[Item]:
    result = await db.execute(select(Item).where(Item.id == item_id))
    return result.scalars().first()

async def get_items(db, skip: int = 0, limit: int = 100) -> List[Item]:
    result = await db.execute(select(Item).offset(skip).limit(limit))
    return result.scalars().all()

async def create_item(db, item: ItemCreate) -> Item:
    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 ValueError("Item with this name already exists")

async def update_item(db, item_id: int, item: ItemUpdate) -> Optional[Item]:
    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, item_id: int) -> bool:
    stmt = delete(Item).where(Item.id == item_id)
    result = await db.execute(stmt)
    await db.commit()
    return result.rowcount > 0

FastAPI Endpoints

Now we expose the CRUD operations through HTTP routes. Each route uses the async dependency get_db to obtain a session.

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

app = FastAPI(title="Async CRUD with FastAPI")

# Create tables on startup
@app.on_event("startup")
async def on_startup():
    async with database.engine.begin() as conn:
        await conn.run_sync(models.Base.metadata.create_all)

@app.post("/items/", response_model=schemas.ItemOut, status_code=status.HTTP_201_CREATED)
async def create_item_endpoint(
    item: schemas.ItemCreate, db: AsyncSession = Depends(database.get_db)
):
    try:
        db_item = await crud.create_item(db, item)
        return db_item
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))

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

@app.get("/items/{item_id}", response_model=schemas.ItemOut)
async def read_item(item_id: int, db: AsyncSession = Depends(database.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.ItemOut)
async def update_item_endpoint(
    item_id: int, item: schemas.ItemUpdate, db: AsyncSession = Depends(database.get_db)
):
    db_item = await crud.update_item(db, item_id, item)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

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

Running the Application

Start the server with the following command:

uvicorn app.main:app --reload

The --reload flag enables hot‑reloading during development. Open http://127.0.0.1:8000/docs to see the automatically generated Swagger UI where you can test each endpoint interactively.

Testing Async CRUD with Pytest

Automated tests guarantee that your API behaves as expected. Below is a minimal async test suite.

# tests/test_items.py
import pytest
from httpx import AsyncClient
from app.main import app
from app.database 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_read_update_delete():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Create
        response = await client.post("/items/", json={"name":"TestItem","price":100})
        assert response.status_code == 201
        item_id = response.json()["id"]

        # Read
        response = await client.get(f"/items/{item_id}")
        assert response.status_code == 200
        assert response.json()["name"] == "TestItem"

        # Update
        response = await client.put(f"/items/{item_id}", json={"price":150})
        assert response.status_code == 200
        assert response.json()["price"] == 150

        # Delete
        response = await client.delete(f"/items/{item_id}")
        assert response.status_code == 204

        # Verify deletion
        response = await client.get(f"/items/{item_id}")
        assert response.status_code == 404

Best Practices for Production‑Ready Async APIs

  • Connection pooling: Configure SQLALCHEMY_POOL_SIZE and MAX_OVERFLOW for PostgreSQL or

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts