sqlalchemy 2.0.51: orm.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.orm.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.orm.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:60c1f6e171817d89cb4926d9bb3ee51ee054ed8f2144d3d24b7b47daa41d4c60","contract":["sessionmaker initializes a Session factory bound to a database engine","sessionmaker instance creates Session objects for executing queries, persisting records, and committing transactions","sessionmaker.begin context manager automatically commits transactions on successful block exit","sessionmaker.begin context manager rolls back transactions when an exception is raised","sessionmaker with expire_on_commit=False keeps object attributes loaded and accessible after transaction commit","sessionmaker.configure dynamically alters factory configuration and binds new engines"],"goal":"verify sqlalchemy.orm.sessionmaker in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.orm.sessionmaker"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.orm.sessionmaker"],"verifierAdapter":"python@1"}

[project]
name = "sample-sqlalchemy-sessionmaker"
version = "0.1.0"
requires-python = ">=3.9"
dependencies = [
    "sqlalchemy==2.0.51",
    "typing-extensions==4.16.0",
]

greenlet==3.5.5
sqlalchemy==2.0.51
typing-extensions==4.16.0

"""Sample demonstrating sqlalchemy.orm.sessionmaker configuration and patterns."""

from typing import Optional, Sequence
from sqlalchemy import Integer, String, create_engine, select
from sqlalchemy.engine import Engine
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

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


def create_sqlite_engine(url: str = "sqlite:///:memory:") -> Engine:
    """Create an SQLite database engine."""
    return create_engine(url, echo=False)


def init_db(engine: Engine) -> None:
    """Initialize schema tables."""
    Base.metadata.create_all(bind=engine)


def create_session_factory(
    engine: Optional[Engine] = None,
    autoflush: bool = True,
    expire_on_commit: bool = True,
) -> sessionmaker[Session]:
    """Create a configured sessionmaker instance."""
    return sessionmaker(
        bind=engine,
        autoflush=autoflush,
        expire_on_commit=expire_on_commit,
    )


def add_user_with_session(session_factory: sessionmaker[Session], name: str, email: str) -> int:
    """Insert a user record using a manually managed session."""
    session = session_factory()
    try:
        user = User(name=name, email=email)
        session.add(user)
        session.commit()
        return user.id
    finally:
        session.close()


def add_user_in_transaction(session_factory: sessionmaker[Session], name: str, email: str) -> User:
    """Insert a user record using sessionmaker.begin() transactional context manager."""
    with session_factory.begin() as session:
        user = User(name=name, email=email)
        session.add(user)
        # Flush is called, and transaction commits upon exiting context
        return user


def get_user_by_id(session_factory: sessionmaker[Session], user_id: int) -> Optional[User]:
    """Query a user by ID."""
    with session_factory() as session:
        return session.get(User, user_id)


def list_users(session_factory: sessionmaker[Session]) -> Sequence[User]:
    """Query all users."""
    with session_factory() as session:
        stmt = select(User).order_by(User.id)
        return session.scalars(stmt).all()

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

"""Contract test proving behaviors of sqlalchemy.orm.sessionmaker in SQLAlchemy 2.0."""

import sys
from pathlib import Path

# Add project root to sys.path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

from sample import (
    Base,
    User,
    add_user_in_transaction,
    add_user_with_session,
    create_session_factory,
    create_sqlite_engine,
    get_user_by_id,
    init_db,
    list_users,
)


def test_sessionmaker_creation_and_bind() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    SessionLocal = sessionmaker(bind=engine)

    assert isinstance(SessionLocal, sessionmaker)
    session = SessionLocal()
    try:
        assert isinstance(session, Session)
        assert session.get_bind() is engine
    finally:
        session.close()


def test_session_crud_and_commit() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    init_db(engine)
    SessionLocal = sessionmaker(bind=engine)

    user_id = add_user_with_session(SessionLocal, "Alice", "alice@example.com")
    assert user_id > 0

    user = get_user_by_id(SessionLocal, user_id)
    assert user is not None
    assert user.name == "Alice"
    assert user.email == "alice@example.com"


def test_sessionmaker_begin_context_manager_commit() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    init_db(engine)
    SessionLocal = sessionmaker(bind=engine)

    with SessionLocal.begin() as session:
        user = User(name="Bob", email="bob@example.com")
        session.add(user)

    users = list_users(SessionLocal)
    names = [u.name for u in users]
    assert "Bob" in names


def test_sessionmaker_begin_context_manager_rollback() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    init_db(engine)
    SessionLocal = sessionmaker(bind=engine)

    try:
        with SessionLocal.begin() as session:
            session.add(User(name="Charlie", email="charlie@example.com"))
            raise RuntimeError("Trigger rollback")
    except RuntimeError:
        pass

    users = list_users(SessionLocal)
    names = [u.name for u in users]
    assert "Charlie" not in names


def test_sessionmaker_expire_on_commit_false() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    init_db(engine)
    SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

    with SessionLocal() as session:
        user = User(name="Dave", email="dave@example.com")
        session.add(user)
        session.commit()

    # Even after session context exits, object attributes remain accessible because expire_on_commit=False
    assert user.id is not None
    assert user.name == "Dave"
    assert user.email == "dave@example.com"


def test_sessionmaker_configure() -> None:
    engine = create_sqlite_engine("sqlite:///:memory:")
    SessionLocal = sessionmaker()

    # Dynamically bind engine using configure()
    SessionLocal.configure(bind=engine)
    session = SessionLocal()
    try:
        assert session.get_bind() is engine
    finally:
        session.close()


def main() -> None:
    test_sessionmaker_creation_and_bind()
    test_session_crud_and_commit()
    test_sessionmaker_begin_context_manager_commit()
    test_sessionmaker_begin_context_manager_rollback()
    test_sessionmaker_expire_on_commit_false()
    test_sessionmaker_configure()
    print("ALL CONTRACT TESTS PASSED")


if __name__ == "__main__":
    main()