sqlalchemy 2.0.51: Integer

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

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:1a2b0a669d4c29320ae703f2731bf0c3cbb521ed8fc45f8c019850cdd82758b3","contract":["sqlalchemy.Integer defines integer column types with python_type int and primary key / autoincrement support","sqlalchemy.Integer maps Python int and None values in ORM mapped models and roundtrips them via Session","sqlalchemy.Integer supports SQL arithmetic, range comparisons, and aggregate functions (sum, count, max)"],"goal":"verify sqlalchemy.Integer in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.Integer"]},"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","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.Integer"],"verifierAdapter":"python@1"}

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

"""SQLAlchemy 2.0 Integer column type, persistence, and query demonstration."""

from typing import Optional, List, Tuple
from sqlalchemy import (
    Engine,
    Integer,
    String,
    create_engine,
    select,
    func,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class InventoryItem(Base):
    __tablename__ = "inventory_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    quantity: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    rating: Mapped[Optional[int]] = mapped_column(Integer, default=None, nullable=True)

    def __repr__(self) -> str:
        return f"<InventoryItem(id={self.id}, name='{self.name}', quantity={self.quantity}, rating={self.rating})>"


def create_database(db_url: str = "sqlite:///:memory:") -> Engine:
    """Create database engine and initialize tables."""
    engine = create_engine(db_url, echo=False)
    Base.metadata.create_all(engine)
    return engine


def add_item(
    session: Session,
    name: str,
    quantity: int = 0,
    rating: Optional[int] = None,
) -> InventoryItem:
    """Add a new inventory item."""
    item = InventoryItem(name=name, quantity=quantity, rating=rating)
    session.add(item)
    session.commit()
    session.refresh(item)
    return item


def update_item_quantity(
    session: Session,
    item_id: int,
    delta: int,
) -> Optional[InventoryItem]:
    """Update quantity of an existing item by delta."""
    stmt = select(InventoryItem).where(InventoryItem.id == item_id)
    item = session.scalars(stmt).one_or_none()
    if item is not None:
        item.quantity += delta
        session.commit()
        session.refresh(item)
    return item


def get_items_by_min_quantity(
    session: Session,
    min_quantity: int,
) -> List[InventoryItem]:
    """Query items with quantity greater than or equal to min_quantity."""
    stmt = (
        select(InventoryItem)
        .where(InventoryItem.quantity >= min_quantity)
        .order_by(InventoryItem.quantity.desc())
    )
    return list(session.scalars(stmt).all())


def get_quantity_aggregates(session: Session) -> Tuple[int, int, Optional[int]]:
    """Compute aggregate statistics: total quantity sum, total item count, and max rating."""
    stmt = select(
        func.sum(InventoryItem.quantity),
        func.count(InventoryItem.id),
        func.max(InventoryItem.rating),
    )
    total_qty, count_id, max_rating = session.execute(stmt).one()
    return int(total_qty or 0), int(count_id), max_rating

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

"""Contract verification test for sqlalchemy.Integer in SQLAlchemy 2.0."""

import os
import sys

# Add sample root directory to sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy import Integer, select
from sqlalchemy.types import TypeEngine
from sqlalchemy.orm import Session
from sample import (
    InventoryItem,
    create_database,
    add_item,
    update_item_quantity,
    get_items_by_min_quantity,
    get_quantity_aggregates,
)


def test_integer_column_metadata() -> None:
    """Assertion 1: sqlalchemy.Integer defines integer column types with python_type int and primary key / autoincrement support."""
    int_type = Integer()
    assert isinstance(int_type, TypeEngine), "Integer must be an instance of TypeEngine"
    assert int_type.python_type is int, f"Expected python_type int, got {int_type.python_type}"

    columns = InventoryItem.__table__.columns

    assert "id" in columns, "Column 'id' must be present"
    assert isinstance(columns["id"].type, Integer), "Column 'id' type must be Integer"
    assert columns["id"].type.python_type is int, "id python_type must be int"
    assert columns["id"].primary_key is True, "id must be primary key"

    assert "quantity" in columns, "Column 'quantity' must be present"
    assert isinstance(columns["quantity"].type, Integer), "Column 'quantity' type must be Integer"
    assert columns["quantity"].type.python_type is int, "quantity python_type must be int"
    assert columns["quantity"].nullable is False, "quantity must be non-nullable"

    assert "rating" in columns, "Column 'rating' must be present"
    assert isinstance(columns["rating"].type, Integer), "Column 'rating' type must be Integer"
    assert columns["rating"].nullable is True, "rating must be nullable"


def test_integer_persistence_and_roundtrip() -> None:
    """Assertion 2: sqlalchemy.Integer maps Python int and None values in ORM mapped models and roundtrips them via Session."""
    engine = create_database()
    with Session(engine) as session:
        item1 = add_item(session, "Widget", quantity=50, rating=5)
        item2 = add_item(session, "Gadget", quantity=15, rating=None)
        item3 = add_item(session, "Gizmo", quantity=0, rating=3)

        assert item1.id is not None
        assert item2.id is not None
        assert item3.id is not None

        # Verify retrieved values and exact int types
        q1 = session.get(InventoryItem, item1.id)
        assert q1 is not None
        assert q1.quantity == 50
        assert isinstance(q1.quantity, int)
        assert q1.rating == 5
        assert isinstance(q1.rating, int)

        q2 = session.get(InventoryItem, item2.id)
        assert q2 is not None
        assert q2.quantity == 15
        assert q2.rating is None

        q3 = session.get(InventoryItem, item3.id)
        assert q3 is not None
        assert q3.quantity == 0
        assert q3.rating == 3

        # Update integer column and verify persistence
        updated = update_item_quantity(session, item2.id, 10)
        assert updated is not None
        assert updated.quantity == 25

        refreshed = session.get(InventoryItem, item2.id)
        assert refreshed is not None
        assert refreshed.quantity == 25


def test_integer_sql_expressions_and_aggregates() -> None:
    """Assertion 3: sqlalchemy.Integer supports SQL arithmetic, range comparisons, and aggregate functions (sum, count, max)."""
    engine = create_database()
    with Session(engine) as session:
        add_item(session, "Item Alpha", quantity=100, rating=5)
        add_item(session, "Item Beta", quantity=40, rating=4)
        add_item(session, "Item Gamma", quantity=10, rating=2)
        add_item(session, "Item Delta", quantity=5, rating=None)

        # Range query (>= min_quantity)
        high_stock = get_items_by_min_quantity(session, 40)
        assert len(high_stock) == 2
        assert [i.name for i in high_stock] == ["Item Alpha", "Item Beta"]

        # Direct SQL arithmetic comparison in where clause (quantity + 5 > 40)
        stmt_math = (
            select(InventoryItem)
            .where((InventoryItem.quantity + 5) > 40)
            .order_by(InventoryItem.quantity)
        )
        math_results = list(session.scalars(stmt_math).all())
        assert len(math_results) == 2
        assert [i.name for i in math_results] == ["Item Beta", "Item Alpha"]

        # Aggregate functions
        total_qty, count_items, max_rate = get_quantity_aggregates(session)
        assert total_qty == 155
        assert count_items == 4
        assert max_rate == 5


def main() -> None:
    test_integer_column_metadata()
    test_integer_persistence_and_roundtrip()
    test_integer_sql_expressions_and_aggregates()
    print("All sqlalchemy.Integer contract assertions passed successfully.")


if __name__ == "__main__":
    main()