arunseohub.in

Python Fastapi Async Crud Operations Tutorial

Written by

arun

in

Welcome to the ultimate Python FastAPI async CRUD operations tutorial! Whether you’re a seasoned developer looking to modernize your API stack or a newcomer eager to build high‑performance web services, FastAPI’s asynchronous capabilities make it the perfect choice. In this guide, we’ll walk through everything you need to know—from setting up a FastAPI project to implementing fully asynchronous create, read, update, and delete (CRUD) endpoints backed by an async‑compatible database. By the end, you’ll have a production‑ready template that you can extend for any real‑world application.

Why Choose FastAPI for Async CRUD?

FastAPI has quickly become the go‑to framework for building RESTful APIs in Python, and for good reasons:

  • Native async support: Built on top of Starlette, FastAPI lets you write async def endpoints that run concurrently, dramatically improving throughput.
  • Automatic OpenAPI documentation: Swagger UI and ReDoc are generated out of the box, making your API self‑documenting.
  • Type‑safety and validation: Pydantic models enforce data schemas, catching errors before they hit your database.
  • Performance: Benchmarks show FastAPI rivals Node.js and Go for request latency.

Prerequisites

Before diving into code, make sure you have the following installed:

  1. Python 3.9 or newer
  2. Git (optional, for version control)
  3. A terminal or command prompt with pip access

We’ll also use uvicorn as the ASGI server and SQLModel (a SQLAlchemy‑based ORM) for async database interactions.

Project Setup

1. Create a virtual environment

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

2. Install dependencies

pip install fastapi uvicorn[standard] sqlmodel aiosqlite

3. Project structure

Organize your files for clarity and scalability:

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

Defining the Data Model

We’ll build a simple Item resource with id, name, description, and price. Using SQLModel gives us both Pydantic validation and SQLAlchemy ORM features.

app/models.py

from sqlmodel import SQLModel, Field
from typing import Optional

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

Creating Pydantic Schemas

Separate schemas for request payloads and response models keep your API clean.

app/schemas.py

from pydantic import BaseModel, Field
from typing import Optional

class ItemCreate(BaseModel):
    name: str = Field(..., max_length=100)
    description: Optional[str] = Field(None, max_length=255)
    price: float = Field(..., gt=0)

class ItemRead(BaseModel):
    id: int
    name: str
    description: Optional[str]
    price: float

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

Async CRUD Functions

All database interactions are async, leveraging SQLModel’s async engine.

app/crud.py

from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from .models import Item
from typing import List, Optional

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

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

async def create_item(session: AsyncSession, item_data) -> Item:
    db_item = Item.from_orm(item_data)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

async def update_item(session: AsyncSession, db_item: Item, updates) -> Item:
    item_data = updates.dict(exclude_unset=True)
    for key, value in item_data.items():
        setattr(db_item, key, value)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

async def delete_item(session: AsyncSession, db_item: Item) -> None:
    await session.delete(db_item)
    await session.commit()

FastAPI Application Entry Point

app/main.py

import uvicorn
from fastapi import FastAPI, HTTPException, Depends, status
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel.ext.asyncio.engine import create_async_engine
from .models import Item
from .schemas import ItemCreate, ItemRead, ItemUpdate
from . import crud

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)

app = FastAPI(
    title="FastAPI Async CRUD Tutorial",
    description="A step‑by‑step guide to building async CRUD endpoints with FastAPI and SQLModel.",
    version="1.0.0"
)

# Dependency to get async DB session
async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

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

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

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

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

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

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

Testing the API with HTTPie or cURL

Once the server is running (uvicorn app.main:app --reload), you can interact with the endpoints:

  • Create an item: 
    http POST http://localhost:8000/items/ name="FastAPI Book" price:=29.99 description="Learn async FastAPI"
  • Read all items: 
    curl -X GET "http://localhost:8000/items/"
  • Update an item: 
    http PATCH http://localhost:8000/items/1 price:=24.99
  • Delete an item: 
    curl -X DELETE "http://localhost:8000/items/1"

Best Practices for Production‑Ready Async CRUD APIs

Building a functional prototype is just the first step. To ensure reliability and scalability, consider the following recommendations:

1. Use a robust database engine

SQLite is great for demos, but for production use PostgreSQL or MySQL with async drivers (asyncpg, aiomysql).

2. Implement pagination and filtering

Instead of returning all rows, add query parameters for limit, offset, and field‑based filters. This reduces memory usage and improves response times.

Comments

Leave a Reply

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

More posts