FastAPI with async SQLAlchemy 2 on aiosqlite, tested offline with per-test in-memory DBs: working versions (tested 2026-10-15)

fastapi==0.143.1
sqlalchemy==2.1.4
aiosqlite==0.22.1
httpx==0.28.1
pytest-asyncio==1.4.0
pytest==9.1.1
greenlet==3.2.4

from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

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


class Base(DeclarativeBase):
    pass


class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    price_cents: Mapped[int]


engine = create_async_engine(DATABASE_URL)
# expire_on_commit=False keeps attributes loaded after commit, so no implicit
# lazy load (which raises MissingGreenlet in async code) when building responses.
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]


class ItemIn(BaseModel):
    name: str
    price_cents: int


class ItemOut(ItemIn):
    model_config = ConfigDict(from_attributes=True)
    id: int


app = FastAPI()


@app.post("/items", response_model=ItemOut, status_code=201)
async def create_item(data: ItemIn, session: SessionDep):
    item = Item(**data.model_dump())
    session.add(item)
    await session.commit()
    return ItemOut.model_validate(item)


@app.get("/items/{item_id}", response_model=ItemOut)
async def read_item(item_id: int, session: SessionDep):
    item = await session.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="item not found")
    return ItemOut.model_validate(item)

import httpx
import pytest_asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

from app import Base, app, get_session


@pytest_asyncio.fixture
async def client():
    # A fresh in-memory database per test. StaticPool forces a single shared
    # connection, because each new sqlite :memory: connection is its own empty DB.
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=StaticPool,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    maker = async_sessionmaker(engine, expire_on_commit=False)

    async def override_get_session():
        async with maker() as session:
            yield session

    app.dependency_overrides[get_session] = override_get_session
    # ASGITransport does not run the app lifespan, so do not rely on startup hooks.
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

    app.dependency_overrides.clear()
    await engine.dispose()


async def test_create_and_fetch(client):
    r = await client.post("/items", json={"name": "pen", "price_cents": 150})
    assert r.status_code == 201
    item_id = r.json()["id"]

    r = await client.get(f"/items/{item_id}")
    assert r.json() == {"id": item_id, "name": "pen", "price_cents": 150}


async def test_missing_item_is_404(client):
    r = await client.get("/items/999")
    assert r.status_code == 404


async def test_invalid_payload_is_422(client):
    r = await client.post("/items", json={"name": "pen", "price_cents": "lots"})
    assert r.status_code == 422


async def test_each_test_gets_fresh_db(client):
    # With a fresh database per test, the first insert always gets id 1.
    r = await client.post("/items", json={"name": "cup", "price_cents": 900})
    assert r.json()["id"] == 1

[pytest]
pythonpath = .
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function