Async repository pattern with SQLAlchemy 2 AsyncSession on aiosqlite: working versions (tested 2026-10-15)

sqlalchemy==2.1.4
aiosqlite==0.22.1
greenlet==3.2.4
pytest-asyncio==1.4.0
pytest==9.1.1

from sqlalchemy import ForeignKey, String, event
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload
from sqlalchemy import select
from sqlalchemy.pool import StaticPool


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] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="books")


def make_engine(url: str = "sqlite+aiosqlite:///:memory:") -> AsyncEngine:
    # StaticPool keeps one connection, so every session sees the same in-memory database.
    engine = create_async_engine(url, poolclass=StaticPool)

    @event.listens_for(engine.sync_engine, "connect")
    def _enable_fk(dbapi_conn, _record):
        # SQLite ignores foreign keys unless enabled on each connection.
        cursor = dbapi_conn.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    return engine


def make_sessionmaker(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    # expire_on_commit=False keeps loaded attributes readable after commit,
    # so async code does not trigger implicit IO.
    return async_sessionmaker(engine, expire_on_commit=False)


class AuthorRepo:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def add(self, name: str) -> Author:
        author = Author(name=name)
        self.session.add(author)
        await self.session.flush()  # assigns author.id without committing
        return author

    async def add_book(self, author_id: int, title: str) -> Book:
        book = Book(title=title, author_id=author_id)
        self.session.add(book)
        await self.session.flush()
        return book

    async def get_with_books(self, author_id: int) -> Author | None:
        stmt = select(Author).options(selectinload(Author.books)).where(Author.id == author_id)
        return (await self.session.execute(stmt)).scalar_one_or_none()

    async def count(self) -> int:
        return len((await self.session.execute(select(Author))).scalars().all())

import pytest
import pytest_asyncio
from sqlalchemy.exc import IntegrityError, MissingGreenlet, StatementError
from repository import Author, AuthorRepo, Base, make_engine, make_sessionmaker


@pytest_asyncio.fixture
async def engine():
    eng = make_engine()
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    await eng.dispose()


@pytest.fixture
def maker(engine):
    return make_sessionmaker(engine)


@pytest.mark.asyncio
async def test_add_and_fetch_books_with_selectinload(maker):
    async with maker() as s:
        repo = AuthorRepo(s)
        author = await repo.add("Ursula")
        await repo.add_book(author.id, "The Dispossessed")
        await s.commit()
    async with maker() as s:
        found = await AuthorRepo(s).get_with_books(author.id)
        assert [b.title for b in found.books] == ["The Dispossessed"]


@pytest.mark.asyncio
async def test_lazy_load_raises_in_async(maker):
    async with maker() as s:
        author = await AuthorRepo(s).add("Octavia")
        await s.commit()
    async with maker() as s:
        loaded = await s.get(Author, author.id)
        # Lazy loading needs IO, which is impossible inside a sync attribute access.
        # SQLAlchemy wraps the MissingGreenlet in a StatementError, so check the original.
        with pytest.raises(StatementError) as info:
            _ = loaded.books
        assert isinstance(info.value.orig, MissingGreenlet)


@pytest.mark.asyncio
async def test_foreign_key_enforced(maker):
    async with maker() as s:
        repo = AuthorRepo(s)
        with pytest.raises(IntegrityError):
            await repo.add_book(999, "Orphan")


@pytest.mark.asyncio
async def test_unique_name_raises(maker):
    async with maker() as s:
        await AuthorRepo(s).add("Dup")
        await s.commit()
    async with maker() as s:
        with pytest.raises(IntegrityError):
            await AuthorRepo(s).add("Dup")


@pytest.mark.asyncio
async def test_rollback_discards_changes(maker):
    async with maker() as s:
        await AuthorRepo(s).add("Ghost")
        await s.rollback()
    async with maker() as s:
        assert await AuthorRepo(s).count() == 0


@pytest.mark.asyncio
async def test_attributes_readable_after_commit(maker):
    async with maker() as s:
        author = await AuthorRepo(s).add("Le Guin")
        await s.commit()
        # expire_on_commit=False: no lazy refresh, so no await is needed here.
        assert author.name == "Le Guin"

[pytest]
pythonpath = .
asyncio_default_fixture_loop_scope = function