#10 Async Database with SQLModel

Module 10 — Async Database with SQLModel

In this module, we learn how to use SQLModel (from FastAPI's creator) to build a fully asynchronous database layer.

---

1️⃣ Install Dependencies

pip install sqlmodel aiosqlite

---

2️⃣ Create Database Models

We create a SQLModel model called Product.

from sqlmodel import SQLModel, Field
from typing import Optional

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    price: float

---

3️⃣ Configure Async Engine & Session

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker

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

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

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

---

4️⃣ FastAPI Startup Event (Create Tables)

from fastapi import FastAPI
from database import init_db
from routes import router

app = FastAPI()

@app.on_event("startup")
async def on_startup():
    await init_db()

app.include_router(router)

---

5️⃣ CRUD Operations (Async)

We create full async CRUD in a router file.

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from database import async_session
from models import Product

router = APIRouter()

async def get_session():
    async with async_session() as session:
        yield session

@router.post("/products", response_model=Product)
async def create_product(product: Product, session: AsyncSession = Depends(get_session)):
    session.add(product)
    await session.commit()
    await session.refresh(product)
    return product

@router.get("/products")
async def list_products(session: AsyncSession = Depends(get_session)):
    result = await session.exec(select(Product))
    return result.all()

@router.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.exec(select(Product).where(Product.id == product_id))
    product = result.first()
    if not product:
        raise HTTPException(404, "Product not found")
    return product

@router.put("/products/{product_id}", response_model=Product)
async def update_product(product_id: int, updated: Product, session: AsyncSession = Depends(get_session)):
    result = await session.exec(select(Product).where(Product.id == product_id))
    product = result.first()
    if not product:
        raise HTTPException(404, "Product not found")

    product.name = updated.name
    product.price = updated.price

    session.add(product)
    await session.commit()
    await session.refresh(product)
    return product

@router.delete("/products/{product_id}")
async def delete_product(product_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.exec(select(Product).where(Product.id == product_id))
    product = result.first()
    if not product:
        raise HTTPException(404, "Product not found")

    await session.delete(product)
    await session.commit()
    return {"status": "deleted"}

---

6️⃣ Run the Server

uvicorn main:app --reload

---

7️⃣ Test CRUD in Swagger UI

Go to: 👉 http://127.0.0.1:8000/docs Try: - ➕ Create product - 📄 List products - 🔍 Get product - ✏️ Update product - ❌ Delete product