sqlalchemy 2.0.51: ext.asyncio.async_sessionmaker

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.async_sessionmaker 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.async_sessionmaker

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:d39f2e23037614e17d8b53355e4918097a897286f16e07044f5bcce629d7c26a","contract":["async_sessionmaker creates AsyncSession instances configured with bind and custom session options","async_sessionmaker.configure dynamically updates default configuration for future sessions","async_sessionmaker supports custom AsyncSession subclasses via class_ parameter","async_sessionmaker instances function as async context managers providing managed AsyncSession lifecycles","async_sessionmaker.begin creates an async context manager managing session transaction lifecycle"],"goal":"verify sqlalchemy.ext.asyncio.async_sessionmaker in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.ext.asyncio.async_sessionmaker"]},"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.async_sessionmaker"],"verifierAdapter":"python@1"}

[project]
name = "verify-sqlalchemy-async-sessionmaker"
version = "0.1.0"
dependencies = [
    "sqlalchemy==2.0.51",
]

sqlalchemy==2.0.51

"""Demonstration of sqlalchemy.ext.asyncio.async_sessionmaker."""

from typing import Any, Optional, Type
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
)


def create_session_factory(
    bind: Optional[Any] = None,
    expire_on_commit: bool = False,
    autoflush: bool = True,
    class_: Type[AsyncSession] = AsyncSession,
) -> async_sessionmaker[AsyncSession]:
    """Create and configure an async_sessionmaker factory."""
    return async_sessionmaker(
        bind=bind,
        expire_on_commit=expire_on_commit,
        autoflush=autoflush,
        class_=class_,
    )

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

"""Contract verification test for sqlalchemy.ext.asyncio.async_sessionmaker."""

import asyncio
import os
import sys
from unittest.mock import MagicMock

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
)
from sample import create_session_factory


class CustomAsyncSession(AsyncSession):
    """Custom AsyncSession subclass for testing session factory class customization."""

    def custom_method(self) -> str:
        return "custom_session_behavior"


def test_async_sessionmaker_instantiation_and_options() -> None:
    """Assertion 1: async_sessionmaker creates AsyncSession instances configured with bind and custom session options."""
    mock_bind = MagicMock()
    factory = create_session_factory(
        bind=mock_bind,
        expire_on_commit=False,
        autoflush=False,
    )

    session = factory()
    assert isinstance(session, AsyncSession), "factory() should return an AsyncSession instance"
    assert session.bind is mock_bind, "session.bind should match the factory bind"
    assert session.sync_session.expire_on_commit is False, "session.sync_session.expire_on_commit should be False"
    assert session.sync_session.autoflush is False, "session.sync_session.autoflush should be False"
    assert session.is_active is True, "session should be active"


def test_async_sessionmaker_configure() -> None:
    """Assertion 2: async_sessionmaker.configure dynamically updates default configuration for future sessions."""
    mock_bind1 = MagicMock()
    mock_bind2 = MagicMock()

    factory = async_sessionmaker(bind=mock_bind1, expire_on_commit=True)
    session1 = factory()
    assert session1.bind is mock_bind1
    assert session1.sync_session.expire_on_commit is True

    # Reconfigure factory
    factory.configure(bind=mock_bind2, expire_on_commit=False)
    session2 = factory()
    assert session2.bind is mock_bind2
    assert session2.sync_session.expire_on_commit is False


def test_async_sessionmaker_custom_class() -> None:
    """Assertion 3: async_sessionmaker supports custom AsyncSession subclasses via class_ parameter."""
    factory = create_session_factory(class_=CustomAsyncSession)
    session = factory()

    assert isinstance(session, CustomAsyncSession), "session should be an instance of CustomAsyncSession"
    assert isinstance(session, AsyncSession), "session should also be an instance of AsyncSession"
    assert session.custom_method() == "custom_session_behavior"


async def async_test_context_manager() -> None:
    """Assertion 4: async_sessionmaker instances function as async context managers providing managed AsyncSession lifecycles."""
    factory = async_sessionmaker(expire_on_commit=False)

    session_ref = None
    async with factory() as session:
        session_ref = session
        assert isinstance(session, AsyncSession)
        assert session.is_active is True

    assert session_ref is not None


async def async_test_begin_context_manager() -> None:
    """Assertion 5: async_sessionmaker.begin creates an async context manager managing session transaction lifecycle."""
    mock_bind = MagicMock()
    factory = async_sessionmaker(bind=mock_bind)

    async with factory.begin() as session:
        assert isinstance(session, AsyncSession)
        assert session.is_active is True
        assert session.in_transaction() is True


def main() -> None:
    test_async_sessionmaker_instantiation_and_options()
    test_async_sessionmaker_configure()
    test_async_sessionmaker_custom_class()
    asyncio.run(async_test_context_manager())
    asyncio.run(async_test_begin_context_manager())
    print("All contract assertions passed.")


if __name__ == "__main__":
    main()