sqlalchemy 2.0.51: text

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.text 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.text
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:393ffdf636d3ba51f9e8428a5575aaaf3d75b0b3f9df9a21f7847db7934574ff","contract":["sqlalchemy.text constructs a TextClause representing raw textual SQL with bound parameter support","sqlalchemy.text supports bound parameters passed directly via execution mapping or through TextClause.bindparams","sqlalchemy.text integrates with Core and ORM select constructs in where and order_by clauses","sqlalchemy.text columns method maps textual SQL result column datatypes to SQLAlchemy types","sqlalchemy.text clause is required for executable raw SQL strings in SQLAlchemy 2.0 while plain strings are rejected"],"goal":"verify sqlalchemy.text in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.text"]},"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.text"],"verifierAdapter":"python@1"}

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

"""SQLAlchemy 2.0 sqlalchemy.text construct usage and verification sample."""

from typing import Any, Dict, List, Optional, Sequence
import sqlalchemy as sa
from sqlalchemy import Float, Integer, String, create_engine, select, text
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    category: Mapped[str] = mapped_column(String(50), nullable=False)
    price: Mapped[float] = mapped_column(Float, nullable=False)

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


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


def insert_products_batch(
    engine: sa.Engine, products_data: Sequence[Dict[str, Any]]
) -> None:
    """Insert multiple products using a text clause with parameter binding."""
    stmt = text(
        "INSERT INTO products (name, category, price) "
        "VALUES (:name, :category, :price)"
    )
    with engine.begin() as conn:
        conn.execute(stmt, products_data)


def find_products_above_price(
    session: Session, min_price: float, order_desc: bool = True
) -> List[Product]:
    """Query products filtering and sorting with text clauses."""
    order_direction = "DESC" if order_desc else "ASC"
    stmt = (
        select(Product)
        .where(text("price >= :min_price"))
        .order_by(text(f"price {order_direction}"))
    )
    return list(session.scalars(stmt, {"min_price": min_price}).all())


def fetch_product_by_name_typed(
    engine: sa.Engine, product_name: str
) -> Optional[Dict[str, Any]]:
    """Execute text query declaring column types via .columns()."""
    stmt = text(
        "SELECT id, name, category, price FROM products WHERE name = :name"
    ).columns(
        id=Integer,
        name=String,
        category=String,
        price=Float,
    )
    with engine.connect() as conn:
        row = conn.execute(stmt, {"name": product_name}).first()
        if row is None:
            return None
        return {
            "id": row.id,
            "name": row.name,
            "category": row.category,
            "price": row.price,
        }


def calculate_category_stats(
    engine: sa.Engine, category_name: str
) -> Dict[str, Any]:
    """Execute parameterized aggregation query using bound parameters."""
    stmt = text(
        "SELECT COUNT(*) AS count, AVG(price) AS avg_price "
        "FROM products WHERE category = :cat"
    ).bindparams(cat=category_name)
    with engine.connect() as conn:
        row = conn.execute(stmt).mappings().first()
        if row is None:
            return {"count": 0, "avg_price": 0.0}
        return {"count": row["count"], "avg_price": row["avg_price"]}


if __name__ == "__main__":
    engine = create_database()
    sample_data = [
        {"name": "Keyboard", "category": "Electronics", "price": 49.99},
        {"name": "Mouse", "category": "Electronics", "price": 24.50},
        {"name": "Notebook", "category": "Stationery", "price": 5.00},
    ]
    insert_products_batch(engine, sample_data)
    with Session(engine) as session:
        items = find_products_above_price(session, 10.0)
        print(f"Products above 10.0: {items}")
    stats = calculate_category_stats(engine, "Electronics")
    print(f"Electronics stats: {stats}")

{
  "schemaVersion": 1,
  "goal": "verify sqlalchemy.text in pkg:pypi/sqlalchemy@2.0.51",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/sqlalchemy@2.0.51"
  ],
  "symbols": [
    "sqlalchemy.text"
  ],
  "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.text in SQLAlchemy 2.0."""

import os
import sys

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

import sqlalchemy as sa
from sqlalchemy import Float, Integer, String, select, text
from sqlalchemy.orm import Session
from sqlalchemy.sql.elements import TextClause

from sample import (
    Product,
    calculate_category_stats,
    create_database,
    fetch_product_by_name_typed,
    find_products_above_price,
    insert_products_batch,
)


def test_text_clause_creation() -> None:
    """sqlalchemy.text constructs a TextClause representing raw textual SQL with bound parameter support"""
    t = text("SELECT :val AS result")
    assert isinstance(t, TextClause)
    assert callable(sa.text)
    assert str(t) == "SELECT :val AS result"
    assert t.text == "SELECT :val AS result"


def test_bound_parameters() -> None:
    """sqlalchemy.text supports bound parameters passed directly via execution mapping or through TextClause.bindparams"""
    engine = create_database()
    with engine.connect() as conn:
        # Scalar evaluation with bound parameter mapping dict
        eval_stmt = text("SELECT :x + :y AS total")
        res = conn.execute(eval_stmt, {"x": 10, "y": 25}).scalar()
        assert res == 35

        # TextClause with pre-bound parameter using .bindparams()
        bound_stmt = text(
            "SELECT :greeting || ', ' || :name || '!' AS msg"
        ).bindparams(greeting="Hello")
        msg = conn.execute(bound_stmt, {"name": "World"}).scalar()
        assert msg == "Hello, World!"

    # Multi-parameter batch execution
    data = [
        {"name": "Monitor", "category": "Electronics", "price": 299.99},
        {"name": "Headphones", "category": "Electronics", "price": 79.99},
        {"name": "Pen", "category": "Stationery", "price": 1.50},
    ]
    insert_products_batch(engine, data)

    stats = calculate_category_stats(engine, "Electronics")
    assert stats["count"] == 2
    assert abs(stats["avg_price"] - 189.99) < 0.01


def test_integration_with_select() -> None:
    """sqlalchemy.text integrates with Core and ORM select constructs in where and order_by clauses"""
    engine = create_database()
    data = [
        {"name": "Item A", "category": "Misc", "price": 10.0},
        {"name": "Item B", "category": "Misc", "price": 30.0},
        {"name": "Item C", "category": "Misc", "price": 20.0},
    ]
    insert_products_batch(engine, data)

    with Session(engine) as session:
        # Filtering with text() inside select().where()
        filtered = find_products_above_price(session, 15.0, order_desc=True)
        assert len(filtered) == 2
        assert filtered[0].name == "Item B"
        assert filtered[0].price == 30.0
        assert filtered[1].name == "Item C"
        assert filtered[1].price == 20.0

        # Ascending order
        filtered_asc = find_products_above_price(session, 15.0, order_desc=False)
        assert filtered_asc[0].name == "Item C"
        assert filtered_asc[1].name == "Item B"


def test_columns_typing() -> None:
    """sqlalchemy.text columns method maps textual SQL result column datatypes to SQLAlchemy types"""
    engine = create_database()
    data = [
        {"name": "Desk Lamp", "category": "Furniture", "price": 35.50},
    ]
    insert_products_batch(engine, data)

    product = fetch_product_by_name_typed(engine, "Desk Lamp")
    assert product is not None
    assert isinstance(product["id"], int)
    assert product["name"] == "Desk Lamp"
    assert isinstance(product["name"], str)
    assert product["category"] == "Furniture"
    assert isinstance(product["category"], str)
    assert isinstance(product["price"], float)
    assert abs(product["price"] - 35.50) < 0.001

    # Non-existent item returns None
    missing = fetch_product_by_name_typed(engine, "Nonexistent")
    assert missing is None


def test_plain_string_rejected() -> None:
    """sqlalchemy.text clause is required for executable raw SQL strings in SQLAlchemy 2.0 while plain strings are rejected"""
    engine = create_database()
    with engine.connect() as conn:
        # Executing a plain string raises ArgumentError or ObjectNotExecutableError in 2.0
        try:
            conn.execute("SELECT 1")  # type: ignore[arg-type]
            raise AssertionError(
                "Passing plain SQL string to execute should raise ArgumentError or ObjectNotExecutableError"
            )
        except (sa.exc.ArgumentError, sa.exc.ObjectNotExecutableError):
            pass

        # Executing via text() succeeds
        result = conn.execute(text("SELECT 1")).scalar()
        assert result == 1


def main() -> None:
    test_text_clause_creation()
    test_bound_parameters()
    test_integration_with_select()
    test_columns_typing()
    test_plain_string_rejected()
    print("All sqlalchemy.text contract tests passed successfully.")


if __name__ == "__main__":
    main()