Skip to content

Async Quickstart (SQLAlchemy 2.x + Pydantic v2)

AsyncBaseViewset mirrors BaseViewset but every CRUD handler is async, backed by an async SQLAlchemy AsyncSession.

Prerequisites

pip install "fastapi-viewsets[sqlalchemy]" aiosqlite

Drivers

For production, use asyncpg (PostgreSQL) or aiomysql (MySQL) instead of aiosqlite.

Configuration

Point SQLALCHEMY_DATABASE_URL (or SQLALCHEMY_ASYNC_DATABASE_URL) at an async-capable URL. The package auto-converts:

Sync URL Async URL
sqlite:/// sqlite+aiosqlite:///
postgresql:// postgresql+asyncpg://
mysql:// mysql+aiomysql://

Create a .env:

ORM_TYPE=sqlalchemy
SQLALCHEMY_DATABASE_URL=sqlite:///./test.db
# Async URL auto-derived; or set explicitly:
# SQLALCHEMY_ASYNC_DATABASE_URL=sqlite+aiosqlite:///./test.db

Full example

Save as main.py in an empty folder and run uvicorn main:app --reload. Open http://127.0.0.1:8000/docs to try the CRUD endpoints.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict
from sqlalchemy import Column, Integer, String
from typing import Optional

from fastapi_viewsets import AsyncBaseViewset
from fastapi_viewsets.db_conf import (
    Base,
    async_engine,
    get_async_session,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create tables once on startup using the async engine.
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(lifespan=lifespan)


class Item(Base):
    """Async-friendly SQLAlchemy model."""

    __tablename__ = "items_async"
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)


class ItemSchema(BaseModel):
    """Pydantic v2 schema reused as request and response model."""

    model_config = ConfigDict(from_attributes=True)
    id: Optional[int] = None
    name: str



items = AsyncBaseViewset(
    endpoint="/items",
    model=Item,
    response_model=ItemSchema,
    db_session=get_async_session,
    tags=["items"],
)
items.register(methods=["LIST", "GET", "POST", "PATCH", "DELETE"])
app.include_router(items)

Key differences from sync

Aspect BaseViewset (sync) AsyncBaseViewset (async)
Handler signature def list(...) async def list(...)
Session type Session AsyncSession
Session factory get_session get_async_session
Table creation Base.metadata.create_all(bind=engine) await conn.run_sync(Base.metadata.create_all)
PATCH semantics model_dump(exclude_unset=True) Same — unset fields preserved

Notes

  • Pydantic v2 is required (pydantic>=2.5). Use model_config = ConfigDict(from_attributes=True) instead of the v1 class Config: orm_mode = True.
  • PATCH uses model_dump(exclude_unset=True) internally, so unset fields are not overwritten with defaults.
  • Missing async driver: if aiosqlite / asyncpg / aiomysql is not installed, sync usage still works — only get_async_session() raises a helpful RuntimeError (since v1.2.1).
  • MSSQL: for SQL Server, set the async URL explicitly via SQLALCHEMY_ASYNC_DATABASE_URL using mssql+aioodbc://....

Next steps