sqlalchemy 2.0.51: orm.InstanceState

# sqlalchemy InstanceState API

This sample pins `sqlalchemy` 2.0.51 and verifies `sqlalchemy.orm.InstanceState` for ORM model state inspection, lifecycle tracking, and attribute history.

The contract verifies:
- `sqlalchemy.inspect` on a mapped Declarative instance returns an `InstanceState` managing identity, session, and attribute state.
- `InstanceState` accurately tracks instance lifecycle states: `transient`, `pending`, `persistent`, and `detached` across `Session` operations.
- `InstanceState` maintains primary key `identity` and `identity_key` once persisted in session.
- `InstanceState` tracks attribute modifications and history via `get_history` and `attrs` collection.
- `InstanceState` tracks attribute expiration and reloads attribute values upon access.

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

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:c73a32a461e7c653f58b914d6c383dbc4aca59d6f5d114012a838f70cc2b9042","contract":["sqlalchemy.inspect on a mapped Declarative instance returns an InstanceState managing identity, session, and attribute state.","InstanceState accurately tracks instance lifecycle states: transient, pending, persistent, and detached across Session operations.","InstanceState maintains primary key identity and identity_key once persisted in session.","InstanceState tracks attribute modifications and history via get_history and attrs collection.","InstanceState tracks attribute expiration and reloads attribute values upon access."],"goal":"verify sqlalchemy.orm.InstanceState in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.orm.InstanceState"]},"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.orm.InstanceState"],"verifierAdapter":"python@1"}

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

"""Demonstration of sqlalchemy.orm.InstanceState inspection and lifecycle management."""

from typing import Optional, Tuple, Any
from sqlalchemy import inspect as sa_inspect
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    InstanceState,
)


class Base(DeclarativeBase):
    pass


class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column()
    count: Mapped[int] = mapped_column(default=0)

    def __repr__(self) -> str:
        return f"<Item(id={self.id}, title='{self.title}', count={self.count})>"


def get_instance_state(instance: Any) -> InstanceState:
    """Return the InstanceState associated with a mapped model instance."""
    state = sa_inspect(instance)
    if not isinstance(state, InstanceState):
        raise TypeError(f"Expected InstanceState, got {type(state)}")
    return state


def inspect_lifecycle(instance: Any) -> dict[str, bool]:
    """Inspect standard lifecycle flags on the given instance."""
    state = get_instance_state(instance)
    return {
        "transient": state.transient,
        "pending": state.pending,
        "persistent": state.persistent,
        "detached": state.detached,
        "deleted": state.deleted,
        "modified": state.modified,
        "has_identity": state.has_identity,
    }


def get_identity_info(instance: Any) -> Tuple[Optional[tuple[Any, ...]], Optional[tuple[Any, ...]]]:
    """Return the identity tuple and identity_key tuple from the instance state."""
    state = get_instance_state(instance)
    return state.identity, state.identity_key

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

"""Contract verification test for sqlalchemy.orm.InstanceState."""

import os
import sys

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

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, InstanceState
from sample import Base, Item, get_instance_state, inspect_lifecycle, get_identity_info


def test_instance_state_acquisition() -> None:
    """Assertion 1: sqlalchemy.inspect on a mapped declarative instance returns an InstanceState."""
    item = Item(title="keyboard", count=1)
    state = get_instance_state(item)

    assert isinstance(state, InstanceState), f"Expected InstanceState, got {type(state)}"
    assert state.object is item, "state.object must reference the mapped instance"
    assert state.class_ is Item, "state.class_ must reference the mapped class"
    assert state.mapper.class_ is Item, "state.mapper must reference the class mapper"


def test_lifecycle_transitions() -> None:
    """Assertion 2: InstanceState accurately tracks instance lifecycle states."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    item = Item(title="mouse", count=2)
    state = get_instance_state(item)

    # 1. Transient
    flags = inspect_lifecycle(item)
    assert flags["transient"] is True
    assert flags["pending"] is False
    assert flags["persistent"] is False
    assert flags["detached"] is False
    assert flags["has_identity"] is False
    assert state.session is None

    # 2. Pending
    with Session(engine) as session:
        session.add(item)
        flags = inspect_lifecycle(item)
        assert flags["transient"] is False
        assert flags["pending"] is True
        assert flags["persistent"] is False
        assert flags["detached"] is False
        assert state.session is session

        # 3. Persistent
        session.flush()
        flags = inspect_lifecycle(item)
        assert flags["pending"] is False
        assert flags["persistent"] is True
        assert flags["detached"] is False
        assert flags["has_identity"] is True

        session.commit()

    # 4. Detached
    flags = inspect_lifecycle(item)
    assert flags["persistent"] is False
    assert flags["detached"] is True
    assert flags["has_identity"] is True
    assert state.session is None


def test_identity_tracking() -> None:
    """Assertion 3: InstanceState maintains primary key identity and identity_key once persisted."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    item = Item(title="monitor", count=3)
    state = get_instance_state(item)
    assert state.identity is None
    assert state.identity_key is None

    with Session(engine) as session:
        session.add(item)
        session.flush()

        identity, identity_key = get_identity_info(item)
        assert identity == (item.id,)
        assert identity_key == (Item, (item.id,), None)
        assert state.key == identity_key
        assert state.has_identity is True


def test_attribute_tracking_and_history() -> None:
    """Assertion 4: InstanceState tracks attribute modifications and history via get_history and attrs."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    item = Item(title="trackpad", count=4)
    with Session(engine) as session:
        session.add(item)
        session.commit()

        state = get_instance_state(item)
        assert state.modified is False

        # Load attribute before modifying so previous value is tracked in history
        _ = item.title
        item.title = "magic trackpad"
        assert state.modified is True

        history = state.get_history("title", passive=False)
        assert history.added == ["magic trackpad"]
        assert history.deleted == ["trackpad"]

        attr_state = state.attrs.title
        assert attr_state.history.added == ["magic trackpad"]
        assert state.committed_state["title"] == "trackpad"

        session.commit()
        assert state.modified is False


def test_attribute_expiration_and_refresh() -> None:
    """Assertion 5: InstanceState tracks attribute expiration and reloads attribute values upon access."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    item = Item(title="desk", count=10)
    with Session(engine) as session:
        session.add(item)
        session.commit()

        state = get_instance_state(item)
        session.expire(item, ["count"])

        assert "count" in state.expired_attributes

        # Accessing expired attribute triggers refresh
        loaded_count = item.count
        assert loaded_count == 10
        assert "count" not in state.expired_attributes


def main() -> None:
    test_instance_state_acquisition()
    test_lifecycle_transitions()
    test_identity_tracking()
    test_attribute_tracking_and_history()
    test_attribute_expiration_and_refresh()
    print("CONTRACT PASS: sqlalchemy.orm.InstanceState verified offline")


if __name__ == "__main__":
    main()