sqlalchemy 2.0.51: engine.reflection.Inspector

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.engine.reflection.Inspector 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.engine.reflection.Inspector
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

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:73ee7f1fe8be2676552f0250285e3db871f32f34d575522816cfc2df14839211","contract":["sqlalchemy.engine.reflection.Inspector inspects schema tables, returning table names and checking table existence.","sqlalchemy.engine.reflection.Inspector inspects column metadata including name, type, and nullability, as well as primary key constraints.","sqlalchemy.engine.reflection.Inspector inspects foreign keys and table indexes, returning referenced tables and index column definitions.","sqlalchemy.engine.reflection.Inspector reflects database table schema into a Table object via reflect_table."],"goal":"verify sqlalchemy.engine.reflection.Inspector in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.engine.reflection.Inspector"]},"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.engine.reflection.Inspector"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51

"""Demonstration of database schema reflection using sqlalchemy.engine.reflection.Inspector."""

from typing import Any, Dict, List, Optional, Union
from sqlalchemy import Connection, Engine, Table, inspect
from sqlalchemy.engine.reflection import Inspector


def get_inspector(bind: Union[Engine, Connection]) -> Inspector:
    """Acquire an Inspector instance from an Engine or Connection."""
    insp = inspect(bind)
    if not isinstance(insp, Inspector):
        raise TypeError(f"Expected Inspector instance, got {type(insp)}")
    return insp


def list_table_names(inspector: Inspector, schema: Optional[str] = None) -> List[str]:
    """Retrieve all table names in the specified or default schema."""
    return inspector.get_table_names(schema=schema)


def check_table_exists(inspector: Inspector, table_name: str, schema: Optional[str] = None) -> bool:
    """Check if a table exists in the database schema."""
    return inspector.has_table(table_name, schema=schema)


def get_table_columns(inspector: Inspector, table_name: str, schema: Optional[str] = None) -> List[Dict[str, Any]]:
    """Inspect column definitions for a specific table."""
    return inspector.get_columns(table_name, schema=schema)


def get_table_primary_key(inspector: Inspector, table_name: str, schema: Optional[str] = None) -> Dict[str, Any]:
    """Inspect primary key constraint information for a table."""
    return inspector.get_pk_constraint(table_name, schema=schema)


def get_table_foreign_keys(inspector: Inspector, table_name: str, schema: Optional[str] = None) -> List[Dict[str, Any]]:
    """Inspect foreign key constraints defined on a table."""
    return inspector.get_foreign_keys(table_name, schema=schema)


def get_table_indexes(inspector: Inspector, table_name: str, schema: Optional[str] = None) -> List[Dict[str, Any]]:
    """Inspect index definitions for a table."""
    return inspector.get_indexes(table_name, schema=schema)


def reflect_table_definition(inspector: Inspector, table: Table) -> None:
    """Reflect table schema and columns into an existing Table object."""
    inspector.reflect_table(table, None)

{
  "schemaVersion": 1,
  "goal": "verify sqlalchemy.engine.reflection.Inspector in pkg:pypi/sqlalchemy@2.0.51",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/sqlalchemy@2.0.51"
  ],
  "symbols": [
    "sqlalchemy.engine.reflection.Inspector"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

"""Contract verification test for sqlalchemy.engine.reflection.Inspector."""

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 import (
    Column,
    ForeignKey,
    Index,
    Integer,
    MetaData,
    String,
    Table,
    UniqueConstraint,
    create_engine,
)
from sqlalchemy.engine.reflection import Inspector
import sample


def create_test_database():
    """Create an in-memory SQLite database populated with test tables and constraints."""
    engine = create_engine("sqlite:///:memory:")
    metadata = MetaData()

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

    posts = Table(
        "posts",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("user_id", Integer, ForeignKey("users.id"), nullable=False),
        Column("title", String(200), nullable=False),
        Index("idx_posts_user_id", "user_id"),
    )

    metadata.create_all(engine)
    return engine


def test_inspector_table_names_and_existence() -> None:
    """Assertion 1: sqlalchemy.engine.reflection.Inspector inspects schema tables, returning table names and checking table existence."""
    engine = create_test_database()
    inspector = sample.get_inspector(engine)

    assert isinstance(inspector, Inspector), f"Expected Inspector instance, got {type(inspector)}"

    tables = sample.list_table_names(inspector)
    assert "users" in tables, f"Expected 'users' in {tables}"
    assert "posts" in tables, f"Expected 'posts' in {tables}"

    assert sample.check_table_exists(inspector, "users") is True
    assert sample.check_table_exists(inspector, "posts") is True
    assert sample.check_table_exists(inspector, "non_existent_table") is False

    with engine.connect() as conn:
        conn_inspector = sample.get_inspector(conn)
        assert isinstance(conn_inspector, Inspector)
        assert conn_inspector.has_table("users") is True


def test_inspector_columns_and_primary_key() -> None:
    """Assertion 2: sqlalchemy.engine.reflection.Inspector inspects column metadata including name, type, and nullability, as well as primary key constraints."""
    engine = create_test_database()
    inspector = sample.get_inspector(engine)

    columns = sample.get_table_columns(inspector, "users")
    col_names = [col["name"] for col in columns]
    assert "id" in col_names
    assert "username" in col_names
    assert "email" in col_names

    col_map = {col["name"]: col for col in columns}
    assert col_map["username"]["nullable"] is False
    assert "VARCHAR" in str(col_map["username"]["type"]).upper()

    pk_info = sample.get_table_primary_key(inspector, "users")
    assert "constrained_columns" in pk_info
    assert pk_info["constrained_columns"] == ["id"]


def test_inspector_foreign_keys_and_indexes() -> None:
    """Assertion 3: sqlalchemy.engine.reflection.Inspector inspects foreign keys and table indexes, returning referenced tables and index column definitions."""
    engine = create_test_database()
    inspector = sample.get_inspector(engine)

    fks = sample.get_table_foreign_keys(inspector, "posts")
    assert len(fks) >= 1
    fk = fks[0]
    assert fk["referred_table"] == "users"
    assert "user_id" in fk["constrained_columns"]
    assert "id" in fk["referred_columns"]

    indexes = sample.get_table_indexes(inspector, "posts")
    idx_names = [idx["name"] for idx in indexes]
    assert "idx_posts_user_id" in idx_names

    post_idx = next(idx for idx in indexes if idx["name"] == "idx_posts_user_id")
    assert "user_id" in post_idx["column_names"]


def test_inspector_reflect_table() -> None:
    """Assertion 4: sqlalchemy.engine.reflection.Inspector reflects database table schema into a Table object via reflect_table."""
    engine = create_test_database()
    inspector = sample.get_inspector(engine)

    target_metadata = MetaData()
    reflected_users = Table("users", target_metadata)
    assert len(reflected_users.columns) == 0

    sample.reflect_table_definition(inspector, reflected_users)

    reflected_col_names = [c.name for c in reflected_users.columns]
    assert "id" in reflected_col_names
    assert "username" in reflected_col_names
    assert "email" in reflected_col_names
    assert len(reflected_users.primary_key.columns) == 1
    assert "id" in [c.name for c in reflected_users.primary_key.columns]


def main() -> None:
    test_inspector_table_names_and_existence()
    test_inspector_columns_and_primary_key()
    test_inspector_foreign_keys_and_indexes()
    test_inspector_reflect_table()
    print("All sqlalchemy.engine.reflection.Inspector contract assertions passed successfully.")


if __name__ == "__main__":
    main()