SQLAlchemy 2 async ORM + aiosqlite with pytest-asyncio: working versions (tested 2026-10-15)
SQLAlchemy 2.x typed async ORM (Mapped, mapped_column, async_sessionmaker, selectinload) on aiosqlite, with pytest-asyncio fixtures.
passed2 passed · 0 failed2026-10-15 11:57 UTC- Status
- passed
- Last passed
- 2026-10-15 11:57 UTC
- Tests
- 2 / 2
- Language
- Python
- Runtime
- Python 3.13.16
uv venv -q .venv && uv pip install -q --python .venv/bin/python -r requirements.txt .venv/bin/python -m pytest -q
Python 3.13.16 linux 6.17.0-1022-azure x64 runner: github-actions 38730348195
The code
Exactly the files that passed. Your agent can pull them with get_recipe("sqlalchemy-async-aiosqlite") (install), or see them on GitHub.
sqlalchemy==2.1.4 aiosqlite==0.22.1 greenlet==3.5.6 pytest-asyncio==1.4.0 pytest==9.1.1
from sqlalchemy import ForeignKey, String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload
class Base(DeclarativeBase):
pass
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100), unique=True)
books: Mapped[list["Book"]] = relationship(back_populates="author", cascade="all, delete-orphan")
class Book(Base):
__tablename__ = "books"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
author: Mapped[Author] = relationship(back_populates="books")
async def make_sessionmaker(url: str = "sqlite+aiosqlite:///:memory:"):
engine = create_async_engine(url)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# expire_on_commit=False avoids lazy-load errors after commit in async code
return engine, async_sessionmaker(engine, expire_on_commit=False)
async def authors_with_books(session: AsyncSession) -> list[Author]:
# async sessions cannot lazy-load; eager-load relationships explicitly
res = await session.scalars(select(Author).options(selectinload(Author.books)).order_by(Author.name))
return list(res)
import pytest
import pytest_asyncio
from sqlalchemy.exc import IntegrityError
from models import Author, Book, authors_with_books, make_sessionmaker
@pytest_asyncio.fixture
async def Session():
engine, sm = await make_sessionmaker()
yield sm
await engine.dispose()
@pytest.mark.asyncio
async def test_create_and_eager_load(Session):
async with Session() as s:
s.add(Author(name="Ada", books=[Book(title="Notes"), Book(title="Engine")]))
await s.commit()
async with Session() as s:
[a] = await authors_with_books(s)
assert sorted(b.title for b in a.books) == ["Engine", "Notes"]
@pytest.mark.asyncio
async def test_unique_violation_rolls_back(Session):
async with Session() as s:
s.add(Author(name="Bo"))
await s.commit()
s.add(Author(name="Bo"))
with pytest.raises(IntegrityError):
await s.commit()
await s.rollback()
assert len(await authors_with_books(s)) == 1
[pytest] pythonpath = .
Connect to ProvenPin
Add the free MCP server so your agent can pull this recipe and others like it. Same setup as the docs.
claude mcp add --transport http provenpin https://mcp.provenpin.com/mcp
{ "mcpServers": { "provenpin": { "url": "https://mcp.provenpin.com/mcp" } } }{ "servers": { "provenpin": { "type": "http", "url": "https://mcp.provenpin.com/mcp" } } }{ "mcpServers": { "provenpin": { "serverUrl": "https://mcp.provenpin.com/mcp" } } }{ "mcpServers": { "provenpin": { "type": "streamableHttp", "url": "https://mcp.provenpin.com/mcp" } } }Any other MCP client: Streamable HTTP at https://mcp.provenpin.com/mcp. No API key, no signup.
Related stacks
Async repository pattern with SQLAlchemy 2 AsyncSession on aiosqlite
A typed async repository over SQLAlchemy 2 AsyncSession with eager loading, foreign keys enforced on SQLite, and offline pytest-asyncio tests against an in-memory database.
FastAPI with async SQLAlchemy 2 on aiosqlite, tested offline with per-test in-memory DBs
A FastAPI app using AsyncSession and async_sessionmaker on aiosqlite, tested with httpx.AsyncClient over ASGITransport, overriding the session dependency with a fresh in-memory database for each test.
Alembic autogenerate compare_metadata drift check against SQLite
Declare SQLAlchemy 2 models with a naming convention, then detect schema drift offline by diffing the live SQLite schema against the metadata with Alembic compare_metadata.
asyncio TaskGroup, timeouts and semaphores tested with pytest-asyncio
Bounded concurrent fetches with asyncio.Semaphore, per-call asyncio.timeout and TaskGroup error handling, tested offline with pytest-asyncio loop scope config and async fixtures.
All SQLAlchemy recipesAll aiosqlite recipesAll pytest-asyncio recipes