sqlalchemy 2.0.51: ext.asyncio.AsyncSession

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify sqlalchemy.ext.asyncio.AsyncSession in pkg:pypi/sqlalchemy@2.0.51
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/sqlalchemy@2.0.51
Demonstrate these symbols/APIs:
  - sqlalchemy.ext.asyncio.AsyncSession

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:6891d89ca3b83d8a54bc9745389f179db4c9a85663c9cc8271fa5b9310225356","contract":["AsyncSession binds to AsyncEngine and manages asynchronous database transactions","AsyncSession tracks object state in new, dirty, and deleted sets across asynchronous CRUD lifecycle","AsyncSession.begin transactional context manager commits on exit and maintains in_transaction status","AsyncSession.refresh reloads expired model instance attributes from the database asynchronously","AsyncSession.execute and AsyncSession.scalars execute queries and return scalar results asynchronously","AsyncSession.run_sync executes synchronous functions passing the underlying Session instance"],"goal":"verify sqlalchemy.ext.asyncio.AsyncSession in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.ext.asyncio.AsyncSession"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.ext.asyncio.AsyncSession"],"verifierAdapter":"python@1"}

[project]
name = "sample-sqlalchemy-asyncsession"
version = "0.1.0"
requires-python = ">=3.9"
dependencies = [
    "aiosqlite==0.20.0",
    "greenlet==3.1.1",
    "sqlalchemy==2.0.51",
    "typing-extensions==4.12.2",
]

aiosqlite==0.20.0
greenlet==3.1.1
sqlalchemy==2.0.51
typing-extensions==4.12.2

"""Sample demonstrating asynchronous database operations with sqlalchemy.ext.asyncio.AsyncSession."""

from typing import Any, Callable, Dict, List, Optional, Sequence, TypeVar
from sqlalchemy import Integer, String, select, text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from sqlalchemy.pool import StaticPool

T = TypeVar("T")


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(100), nullable=False)


def create_sqlite_async_engine(url: str = "sqlite+aiosqlite:///:memory:") -> AsyncEngine:
    """Create an SQLite async database engine with StaticPool for in-memory isolation."""
    return create_async_engine(
        url,
        echo=False,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )


async def init_db(engine: AsyncEngine) -> None:
    """Initialize schema tables asynchronously."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


def create_async_session_factory(
    engine: AsyncEngine,
    expire_on_commit: bool = False,
) -> async_sessionmaker[AsyncSession]:
    """Create a configured async_sessionmaker factory."""
    return async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=expire_on_commit,
    )


async def add_user(session: AsyncSession, username: str, email: str) -> User:
    """Insert a single user record using AsyncSession."""
    user = User(username=username, email=email)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


async def get_user_by_id(session: AsyncSession, user_id: int) -> Optional[User]:
    """Fetch a user record by primary key using AsyncSession.get."""
    return await session.get(User, user_id)


async def list_users(session: AsyncSession) -> Sequence[User]:
    """Query all users using AsyncSession.scalars."""
    stmt = select(User).order_by(User.id)
    result = await session.scalars(stmt)
    return result.all()


async def update_user_email(session: AsyncSession, user_id: int, new_email: str) -> Optional[User]:
    """Update a user's email address and commit asynchronously."""
    user = await session.get(User, user_id)
    if user is not None:
        user.email = new_email
        await session.commit()
        await session.refresh(user)
    return user


async def delete_user(session: AsyncSession, user_id: int) -> bool:
    """Delete a user record and commit asynchronously."""
    user = await session.get(User, user_id)
    if user is not None:
        await session.delete(user)
        await session.commit()
        return True
    return False


async def transactional_batch_insert(
    session: AsyncSession,
    users_data: Sequence[Dict[str, str]],
) -> None:
    """Insert multiple users inside an explicit transaction block."""
    async with session.begin():
        for item in users_data:
            user = User(username=item["username"], email=item["email"])
            session.add(user)


async def run_sync_operation(
    session: AsyncSession,
    fn: Callable[[Session], T],
) -> T:
    """Execute a synchronous operation with the underlying sync session."""
    return await session.run_sync(fn)

{
  "schemaVersion": 1,
  "goal": "verify sqlalchemy.ext.asyncio.AsyncSession in pkg:pypi/sqlalchemy@2.0.51",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/sqlalchemy@2.0.51"
  ],
  "symbols": [
    "sqlalchemy.ext.asyncio.AsyncSession"
  ]
}

"""Contract test verifying sqlalchemy.ext.asyncio.AsyncSession in SQLAlchemy 2.0."""

import asyncio
import os
import sys

# Ensure repository root is on sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from sqlalchemy.orm import Session

from sample import (
    Base,
    User,
    add_user,
    create_async_session_factory,
    create_sqlite_async_engine,
    delete_user,
    get_user_by_id,
    init_db,
    list_users,
    run_sync_operation,
    transactional_batch_insert,
    update_user_email,
)


async def test_async_session_initialization_and_bind() -> None:
    """Assertion 1: AsyncSession binds to AsyncEngine and manages asynchronous database transactions."""
    engine = create_sqlite_async_engine()
    await init_db(engine)

    session = AsyncSession(bind=engine)
    try:
        assert isinstance(session, AsyncSession)
        assert session.bind is engine
        assert session.is_active is True
        assert session.in_transaction() is False
        assert session.sync_session is not None
    finally:
        await session.close()

    await engine.dispose()


async def test_async_session_crud_and_state_tracking() -> None:
    """Assertion 2: AsyncSession tracks object state in new, dirty, and deleted sets across asynchronous CRUD lifecycle."""
    engine = create_sqlite_async_engine()
    await init_db(engine)
    session_factory = create_async_session_factory(engine)

    async with session_factory() as session:
        # Create (Add)
        user = User(username="user1", email="user1@example.com")
        session.add(user)
        assert user in session.new
        await session.commit()
        assert user.id is not None
        user_id = user.id

        # Read (Get)
        fetched = await get_user_by_id(session, user_id)
        assert fetched is not None
        assert fetched.username == "user1"
        assert fetched.email == "user1@example.com"

        # Read (Query / Scalars)
        users = await list_users(session)
        assert len(users) == 1
        assert users[0].username == "user1"

        # Update (Dirty)
        fetched.email = "user1_updated@example.com"
        assert fetched in session.dirty
        await session.commit()
        assert fetched.email == "user1_updated@example.com"

        # Delete (Deleted)
        await session.delete(fetched)
        assert fetched in session.deleted
        await session.commit()

        # Verify deletion
        fetched_after = await get_user_by_id(session, user_id)
        assert fetched_after is None

    await engine.dispose()


async def test_async_session_begin_commit_and_rollback() -> None:
    """Assertion 3: AsyncSession.begin transactional context manager commits on exit and maintains in_transaction status."""
    engine = create_sqlite_async_engine()
    await init_db(engine)
    session_factory = create_async_session_factory(engine)

    # Success scenario: commit upon block exit
    async with session_factory() as session:
        assert session.in_transaction() is False
        items = [
            {"username": "user_tx1", "email": "tx1@example.com"},
            {"username": "user_tx2", "email": "tx2@example.com"},
        ]
        await transactional_batch_insert(session, items)
        assert session.in_transaction() is False

    async with session_factory() as session:
        users = await list_users(session)
        usernames = [u.username for u in users]
        assert "user_tx1" in usernames
        assert "user_tx2" in usernames

    # Failure scenario: rollback on exception
    async with session_factory() as session:
        try:
            async with session.begin():
                assert session.in_transaction() is True
                session.add(User(username="user_tx3", email="tx3@example.com"))
                # Trigger error (duplicate username user_tx1)
                session.add(User(username="user_tx1", email="conflict@example.com"))
        except IntegrityError:
            pass
        assert session.in_transaction() is False

    async with session_factory() as session:
        users = await list_users(session)
        usernames = [u.username for u in users]
        assert "user_tx3" not in usernames

    await engine.dispose()


async def test_async_session_refresh() -> None:
    """Assertion 4: AsyncSession.refresh reloads expired model instance attributes from the database asynchronously."""
    engine = create_sqlite_async_engine()
    await init_db(engine)
    session_factory = create_async_session_factory(engine, expire_on_commit=True)

    async with session_factory() as session:
        user = User(username="refresh_user", email="refresh@example.com")
        session.add(user)
        await session.commit()

        # Attribute access triggers expired attribute refresh
        await session.refresh(user)
        assert user.id is not None
        assert user.username == "refresh_user"
        assert user.email == "refresh@example.com"

    await engine.dispose()


async def test_async_session_execute_and_scalars() -> None:
    """Assertion 5: AsyncSession.execute and AsyncSession.scalars execute queries and return scalar results asynchronously."""
    engine = create_sqlite_async_engine()
    await init_db(engine)
    session_factory = create_async_session_factory(engine)

    async with session_factory() as session:
        session.add_all([
            User(username="scalar_a", email="a@example.com"),
            User(username="scalar_b", email="b@example.com"),
        ])
        await session.commit()

        # execute with Core text query
        result = await session.execute(text("SELECT count(*) FROM users"))
        count = result.scalar_one()
        assert count == 2

        # execute with ORM select query
        stmt = select(User).where(User.username == "scalar_a")
        res = await session.execute(stmt)
        user_a = res.scalar_one()
        assert user_a.username == "scalar_a"

    await engine.dispose()


async def test_async_session_run_sync() -> None:
    """Assertion 6: AsyncSession.run_sync executes synchronous functions passing the underlying Session instance."""
    engine = create_sqlite_async_engine()
    await init_db(engine)
    session_factory = create_async_session_factory(engine)

    async with session_factory() as session:
        def sync_callback(sync_session: Session) -> bool:
            return sync_session.is_active

        is_active = await run_sync_operation(session, sync_callback)
        assert is_active is True

    await engine.dispose()


async def main_async() -> None:
    await test_async_session_initialization_and_bind()
    await test_async_session_crud_and_state_tracking()
    await test_async_session_begin_commit_and_rollback()
    await test_async_session_refresh()
    await test_async_session_execute_and_scalars()
    await test_async_session_run_sync()
    print("All sqlalchemy.ext.asyncio.AsyncSession contract tests passed.")


def main() -> None:
    asyncio.run(main_async())


if __name__ == "__main__":
    main()