sqlalchemy 2.0.51: sqlalchemy

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

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:dab566cf4a0aac4d8f28b05d48ad87e13b5c97a28671903ecb451b8c023fad46","contract":["assert sqlalchemy.create_engine connects to SQLite and executes text statements","assert sqlalchemy Table, Column, Integer, and String define schema and create tables via MetaData","assert sqlalchemy.insert and sqlalchemy.select perform Core SQL operations","assert sqlalchemy.orm DeclarativeBase, Mapped, and mapped_column declare typed ORM models","assert sqlalchemy.orm.Session persists entities, queries with select, and handles rollback","assert sqlalchemy.exc.IntegrityError is raised on unique constraint violation"],"goal":"verify sqlalchemy in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy"]},"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"],"verifierAdapter":"python@1"}

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

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

"""SQLAlchemy 2.0 sample module."""

"""SQLAlchemy 2.0 Core and ORM demonstration."""

from typing import List, Optional
import sqlalchemy as sa
from sqlalchemy import (
    Column,
    ForeignKey,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
    insert,
    select,
    text,
)
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    relationship,
)


# --- Core Schema Definition ---
metadata = MetaData()

users_table = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String(50), nullable=False, unique=True),
    Column("email", String(100), nullable=False),
)


# --- ORM Schema Definition ---
class Base(DeclarativeBase):
    pass


class Department(Base):
    __tablename__ = "departments"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
    employees: Mapped[List["Employee"]] = relationship(
        back_populates="department", cascade="all, delete-orphan"
    )


class Employee(Base):
    __tablename__ = "employees"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    role: Mapped[str] = mapped_column(String(50), nullable=False)
    department_id: Mapped[int] = mapped_column(
        ForeignKey("departments.id"), nullable=False
    )

    department: Mapped[Department] = relationship(back_populates="employees")


def init_engine(db_url: str = "sqlite:///:memory:") -> sa.Engine:
    """Create an SQLite engine and initialize Core and ORM tables."""
    engine = create_engine(db_url, echo=False)
    metadata.create_all(engine)
    Base.metadata.create_all(engine)
    return engine


def core_insert_user(engine: sa.Engine, name: str, email: str) -> int:
    """Insert a user using SQLAlchemy Core insert statement."""
    with engine.begin() as conn:
        stmt = insert(users_table).values(name=name, email=email)
        result = conn.execute(stmt)
        return result.inserted_primary_key[0]


def core_select_users(engine: sa.Engine) -> List[dict]:
    """Select users using SQLAlchemy Core select statement."""
    with engine.connect() as conn:
        stmt = select(users_table).order_by(users_table.c.id)
        result = conn.execute(stmt)
        return [dict(row._mapping) for row in result]


def orm_create_department(
    session: Session, name: str, employee_names: Optional[List[str]] = None
) -> Department:
    """Create a department with optional initial employees."""
    dept = Department(name=name)
    if employee_names:
        for emp_name in employee_names:
            dept.employees.append(Employee(name=emp_name, role="Staff"))
    session.add(dept)
    session.commit()
    session.refresh(dept)
    return dept


def orm_get_department_by_name(session: Session, name: str) -> Optional[Department]:
    """Query a department by name using 2.0 select."""
    stmt = select(Department).where(Department.name == name)
    return session.scalars(stmt).one_or_none()

"""Contract verification test for SQLAlchemy 2.0.51."""

import os
import sys
from pathlib import Path

# Add root directory to sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import sqlalchemy as sa
from sqlalchemy import select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from src.sample import (
    Base,
    Department,
    Employee,
    core_insert_user,
    core_select_users,
    init_engine,
    metadata,
    orm_create_department,
    orm_get_department_by_name,
    users_table,
)


def test_core_engine_and_text() -> None:
    """Assertion: assert sqlalchemy.create_engine connects to SQLite and executes text statements."""
    engine = init_engine()
    with engine.connect() as conn:
        result = conn.execute(text("SELECT 1 + 1 AS sum_val"))
        row = result.one()
        assert row.sum_val == 2
        assert row[0] == 2
        assert row._mapping["sum_val"] == 2


def test_core_table_and_metadata() -> None:
    """Assertion: assert sqlalchemy Table, Column, Integer, and String define schema and create tables via MetaData."""
    assert users_table.name == "users"
    assert "id" in users_table.c
    assert "name" in users_table.c
    assert "email" in users_table.c
    assert users_table.c.id.primary_key is True
    assert users_table.c.name.unique is True


def test_core_crud_operations() -> None:
    """Assertion: assert sqlalchemy.insert and sqlalchemy.select perform Core SQL operations."""
    engine = init_engine()
    u1_id = core_insert_user(engine, "Alice", "alice@example.com")
    u2_id = core_insert_user(engine, "Bob", "bob@example.com")
    assert u1_id == 1
    assert u2_id == 2

    users = core_select_users(engine)
    assert len(users) == 2
    assert users[0] == {"id": 1, "name": "Alice", "email": "alice@example.com"}
    assert users[1] == {"id": 2, "name": "Bob", "email": "bob@example.com"}

    # Test unique constraint on Core table
    try:
        core_insert_user(engine, "Alice", "duplicate@example.com")
        raise AssertionError("Duplicate name should raise IntegrityError")
    except IntegrityError:
        pass


def test_orm_declarative_and_mapping() -> None:
    """Assertion: assert sqlalchemy.orm DeclarativeBase, Mapped, and mapped_column declare typed ORM models."""
    assert issubclass(Department, Base)
    assert issubclass(Employee, Base)
    assert Department.__tablename__ == "departments"
    assert Employee.__tablename__ == "employees"
    assert "id" in Department.__table__.columns
    assert "name" in Department.__table__.columns
    assert "department_id" in Employee.__table__.columns


def test_orm_session_crud_and_rollback() -> None:
    """Assertion: assert sqlalchemy.orm.Session persists entities, queries with select, and handles rollback."""
    engine = init_engine()
    with Session(engine) as session:
        dept = orm_create_department(
            session, "Engineering", ["Developer A", "Developer B"]
        )
        assert dept.id is not None
        assert len(dept.employees) == 2

        # Query using 2.0 select
        queried = orm_get_department_by_name(session, "Engineering")
        assert queried is not None
        assert queried.id == dept.id
        assert len(queried.employees) == 2

        # Test Transaction Rollback
        uncommitted = Department(name="Temporary")
        session.add(uncommitted)
        session.flush()

        temp_check = session.scalars(
            select(Department).where(Department.name == "Temporary")
        ).one_or_none()
        assert temp_check is not None

        session.rollback()

        after_rollback = session.scalars(
            select(Department).where(Department.name == "Temporary")
        ).one_or_none()
        assert after_rollback is None

        # Ensure committed data remains
        persisted = orm_get_department_by_name(session, "Engineering")
        assert persisted is not None


def test_integrity_error_handling() -> None:
    """Assertion: assert sqlalchemy.exc.IntegrityError is raised on unique constraint violation."""
    engine = init_engine()
    with Session(engine) as session:
        orm_create_department(session, "HR")
        duplicate = Department(name="HR")
        session.add(duplicate)
        try:
            session.commit()
            raise AssertionError("Duplicate department name must raise IntegrityError")
        except IntegrityError:
            session.rollback()


def main() -> None:
    print(f"Testing with SQLAlchemy version: {sa.__version__}")
    assert sa.__version__ == "2.0.51", f"Expected version 2.0.51, got {sa.__version__}"

    test_core_engine_and_text()
    test_core_table_and_metadata()
    test_core_crud_operations()
    test_orm_declarative_and_mapping()
    test_orm_session_crud_and_rollback()
    test_integrity_error_handling()

    print("All contract assertions passed successfully.")


if __name__ == "__main__":
    main()