sqlalchemy 2.0.51: inspect

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

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:f53984917619d461b441e45e5659963ce7df2fa9faaace7d3d5b4fb8982bf7e1","contract":["sqlalchemy.inspect reflects database schema from an Engine or Connection, returning an Inspector capable of querying table names and column metadata.","sqlalchemy.inspect inspects mapped ORM classes, returning a Mapper providing access to underlying local_table, primary_key columns, and relationships.","sqlalchemy.inspect inspects ORM model instances, returning an InstanceState tracking object lifecycle states (transient, persistent) and attribute values.","sqlalchemy.inspect raises NoInspectionAvailable when inspecting unsupported types if raiseerr=True, and returns None if raiseerr=False."],"goal":"verify sqlalchemy.inspect in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.inspect"]},"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.inspect"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51

"""Clean demonstration of schema and ORM inspection using sqlalchemy.inspect."""

from typing import Any, List
from sqlalchemy import (
    Column,
    ForeignKey,
    Integer,
    String,
    create_engine,
    inspect,
)
from sqlalchemy.engine import Engine
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.orm import DeclarativeBase, InstanceState, Mapper, relationship


class Base(DeclarativeBase):
    """Base declarative class for ORM models."""


class Department(Base):
    __tablename__ = "departments"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

    employees = relationship("Employee", back_populates="department")


class Employee(Base):
    __tablename__ = "employees"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    department_id = Column(Integer, ForeignKey("departments.id"))

    department = relationship("Department", back_populates="employees")


def init_database() -> Engine:
    """Initialize an in-memory SQLite database with sample schema."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    return engine


def get_inspector(target: Any) -> Inspector:
    """Obtain an Inspector instance for an Engine or Connection using inspect."""
    return inspect(target)


def get_table_names(engine: Engine) -> List[str]:
    """Retrieve all table names via the reflected Inspector."""
    inspector = inspect(engine)
    return inspector.get_table_names()


def get_table_column_names(engine: Engine, table_name: str) -> List[str]:
    """Retrieve column names for a specific table via the reflected Inspector."""
    inspector = inspect(engine)
    return [col["name"] for col in inspector.get_columns(table_name)]


def get_model_mapper(model_cls: type) -> Mapper:
    """Obtain the Mapper instance for a declarative class using inspect."""
    return inspect(model_cls)


def get_model_primary_keys(model_cls: type) -> List[str]:
    """Retrieve primary key column names from a mapped class using inspect."""
    mapper = inspect(model_cls)
    return [col.name for col in mapper.primary_key]


def get_instance_state(instance: Any) -> InstanceState:
    """Obtain the InstanceState tracking object for an ORM instance using inspect."""
    return inspect(instance)


def inspect_safely(target: Any, raise_error: bool = True) -> Any:
    """Inspect target with optional exception suppression."""
    return inspect(target, raiseerr=raise_error)

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

"""Contract verification tests for sqlalchemy.inspect."""

import os
import sys

# Ensure sample root directory is in sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.exc import NoInspectionAvailable
from sqlalchemy.orm import InstanceState, Mapper
import sample


def test_inspect_engine_schema() -> None:
    """Assertion 1: sqlalchemy.inspect reflects database schema from an Engine or Connection, returning an Inspector capable of querying table names and column metadata."""
    engine = sample.init_database()
    inspector = sample.get_inspector(engine)

    assert isinstance(inspector, Inspector)
    tables = sample.get_table_names(engine)
    assert "departments" in tables
    assert "employees" in tables

    emp_columns = sample.get_table_column_names(engine, "employees")
    assert "id" in emp_columns
    assert "name" in emp_columns
    assert "department_id" in emp_columns

    assert inspector.has_table("employees") is True
    assert inspector.has_table("non_existent_table") is False


def test_inspect_mapped_class() -> None:
    """Assertion 2: sqlalchemy.inspect inspects mapped ORM classes, returning a Mapper providing access to underlying local_table, primary_key columns, and relationships."""
    mapper = sample.get_model_mapper(sample.Employee)

    assert isinstance(mapper, Mapper)
    assert mapper.local_table.name == "employees"

    pk_names = sample.get_model_primary_keys(sample.Employee)
    assert pk_names == ["id"]

    assert "department" in mapper.relationships.keys()


def test_inspect_mapped_instance() -> None:
    """Assertion 3: sqlalchemy.inspect inspects ORM model instances, returning an InstanceState tracking object lifecycle states (transient, persistent) and attribute values."""
    emp = sample.Employee(id=1, name="Alice")
    state = sample.get_instance_state(emp)

    assert isinstance(state, InstanceState)
    assert state.transient is True
    assert state.persistent is False
    assert state.attrs.name.loaded_value == "Alice"


def test_inspect_unsupported_targets() -> None:
    """Assertion 4: sqlalchemy.inspect raises NoInspectionAvailable when inspecting unsupported types if raiseerr=True, and returns None if raiseerr=False."""
    try:
        sample.inspect_safely(42, raise_error=True)
        assert False, "Expected NoInspectionAvailable exception"
    except NoInspectionAvailable:
        pass

    result = sample.inspect_safely(42, raise_error=False)
    assert result is None


def main() -> None:
    test_inspect_engine_schema()
    test_inspect_mapped_class()
    test_inspect_mapped_instance()
    test_inspect_unsupported_targets()
    print("All sqlalchemy.inspect contract assertions passed successfully.")


if __name__ == "__main__":
    main()