sqlalchemy 2.0.51: Engine

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

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:b19d590204eb2782baa8d246f2b94a2d7b68d51eccecc55c23f0a4e29f058a9e","contract":["sqlalchemy.create_engine creates an Engine instance configured with connection pool and dialect","sqlalchemy.Engine.connect acquires connections and executes text() SQL statements with explicit transaction control","sqlalchemy.Engine.begin manages transactional execution blocks with automatic commit and rollback","sqlalchemy.Engine in 2.0 does not support connectionless execution (Engine.execute is removed)","sqlalchemy.Engine.dispose closes and releases pooled database connections"],"goal":"verify sqlalchemy.Engine in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.Engine"]},"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"],"verifierAdapter":"python@1"}

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

"""SQLAlchemy 2.0 Engine and connection lifecycle management sample."""

from typing import Any, List, Optional, Sequence, Tuple
from sqlalchemy import Engine, create_engine, text
from sqlalchemy.engine import Connection, CursorResult
from sqlalchemy.pool import Pool


def create_sqlite_engine(db_url: str = "sqlite:///:memory:", echo: bool = False) -> Engine:
    """Create a configured SQLAlchemy Engine instance."""
    return create_engine(db_url, echo=echo)


def init_db(engine: Engine) -> None:
    """Initialize table schema using engine.begin context manager."""
    with engine.begin() as conn:
        conn.execute(
            text(
                "CREATE TABLE records ("
                "  id INTEGER PRIMARY KEY AUTOINCREMENT,"
                "  key TEXT NOT NULL UNIQUE,"
                "  value INTEGER NOT NULL"
                ")"
            )
        )


def insert_record(engine: Engine, key: str, value: int) -> int:
    """Insert a single record within a committed transaction using engine.begin()."""
    with engine.begin() as conn:
        result: CursorResult[Any] = conn.execute(
            text("INSERT INTO records (key, value) VALUES (:key, :value)"),
            {"key": key, "value": value},
        )
        return int(result.rowcount)


def insert_records_batch(engine: Engine, records: Sequence[Tuple[str, int]]) -> int:
    """Insert multiple records in a transaction block."""
    with engine.begin() as conn:
        result = conn.execute(
            text("INSERT INTO records (key, value) VALUES (:key, :value)"),
            [{"key": k, "value": v} for k, v in records],
        )
        return int(result.rowcount)


def fetch_all_records(engine: Engine) -> List[Tuple[int, str, int]]:
    """Fetch all records using engine.connect() context manager."""
    with engine.connect() as conn:
        result = conn.execute(text("SELECT id, key, value FROM records ORDER BY id"))
        return [(int(row[0]), str(row[1]), int(row[2])) for row in result.fetchall()]


def fetch_value_by_key(engine: Engine, key: str) -> Optional[int]:
    """Fetch a scalar value by key using engine.connect()."""
    with engine.connect() as conn:
        result = conn.execute(
            text("SELECT value FROM records WHERE key = :key"),
            {"key": key},
        )
        scalar = result.scalar_one_or_none()
        return int(scalar) if scalar is not None else None


def execute_transaction_with_rollback(engine: Engine, key: str, value: int) -> None:
    """Attempt an operation in a transaction that raises and rolls back."""
    with engine.begin() as conn:
        conn.execute(
            text("INSERT INTO records (key, value) VALUES (:key, :value)"),
            {"key": key, "value": value},
        )
        raise RuntimeError("simulated transaction failure")


def dispose_engine(engine: Engine) -> None:
    """Dispose connection pool resources associated with the Engine."""
    engine.dispose()

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

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

import os
import sys
from pathlib import Path

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import sqlalchemy as sa
from sqlalchemy import Engine, create_engine, text
from sqlalchemy.engine import Connection
from sqlalchemy.exc import ArgumentError, ObjectNotExecutableError
from sqlalchemy.pool import Pool

from sample import (
    create_sqlite_engine,
    dispose_engine,
    execute_transaction_with_rollback,
    fetch_all_records,
    fetch_value_by_key,
    init_db,
    insert_record,
    insert_records_batch,
)


def test_engine_creation_and_properties() -> None:
    """Assertion: sqlalchemy.create_engine creates an Engine instance configured with connection pool and dialect."""
    engine = create_sqlite_engine()
    try:
        assert isinstance(engine, Engine), "Engine instance must be of type sqlalchemy.Engine"
        assert engine.dialect.name == "sqlite", f"Expected sqlite dialect, got {engine.dialect.name}"
        assert engine.url.drivername == "sqlite", f"Expected sqlite drivername, got {engine.url.drivername}"
        assert isinstance(engine.pool, Pool), "Engine.pool must be an instance of sqlalchemy.pool.Pool"
        assert engine.echo is False, "Echo setting should match configuration"
    finally:
        dispose_engine(engine)


def test_engine_connect_and_execution() -> None:
    """Assertion: sqlalchemy.Engine.connect acquires connections and executes text() SQL statements with explicit transaction control."""
    engine = create_sqlite_engine()
    try:
        init_db(engine)

        # Connect and execute valid text() statement
        with engine.connect() as conn:
            assert isinstance(conn, Connection), "engine.connect() must yield a Connection instance"
            scalar_res = conn.execute(text("SELECT 42")).scalar()
            assert scalar_res == 42, f"Expected 42, got {scalar_res}"

            # Bare strings are not executable in SQLAlchemy 2.0
            try:
                conn.execute("SELECT 42")  # type: ignore[arg-type]
                raise AssertionError("Bare string query should raise ObjectNotExecutableError")
            except ObjectNotExecutableError:
                pass

            # Explicit commit-as-you-go within connect()
            conn.execute(
                text("INSERT INTO records (key, value) VALUES (:key, :value)"),
                {"key": "alpha", "value": 100},
            )
            conn.commit()

        # Verify committed record in a new connection
        with engine.connect() as conn:
            val = conn.execute(
                text("SELECT value FROM records WHERE key = :key"),
                {"key": "alpha"},
            ).scalar()
            assert val == 100, f"Expected value 100, got {val}"
    finally:
        dispose_engine(engine)


def test_engine_begin_transaction_lifecycle() -> None:
    """Assertion: sqlalchemy.Engine.begin manages transactional execution blocks with automatic commit and rollback."""
    engine = create_sqlite_engine()
    try:
        init_db(engine)

        # Successful transaction block automatically commits
        insert_record(engine, "item1", 10)
        insert_records_batch(engine, [("item2", 20), ("item3", 30)])

        records = fetch_all_records(engine)
        assert len(records) == 3, f"Expected 3 records, got {len(records)}"
        assert fetch_value_by_key(engine, "item2") == 20

        # Transaction block that encounters an exception automatically rolls back
        try:
            execute_transaction_with_rollback(engine, "rolled_back_item", 999)
            raise AssertionError("execute_transaction_with_rollback should have raised RuntimeError")
        except RuntimeError as exc:
            assert "simulated transaction failure" in str(exc)

        # Verify that rolled_back_item was not persisted
        records_after = fetch_all_records(engine)
        assert len(records_after) == 3, f"Record count should remain 3, got {len(records_after)}"
        assert fetch_value_by_key(engine, "rolled_back_item") is None
    finally:
        dispose_engine(engine)


def test_engine_connectionless_execution_removed() -> None:
    """Assertion: sqlalchemy.Engine in 2.0 does not support connectionless execution (Engine.execute is removed)."""
    engine = create_sqlite_engine()
    try:
        assert not hasattr(engine, "execute"), "Engine.execute must not exist in SQLAlchemy 2.0"
    finally:
        dispose_engine(engine)


def test_engine_dispose_cleanup() -> None:
    """Assertion: sqlalchemy.Engine.dispose closes and releases pooled database connections."""
    engine = create_sqlite_engine()
    init_db(engine)

    # Establish connection
    with engine.connect() as conn:
        res = conn.execute(text("SELECT 1")).scalar()
        assert res == 1

    # Dispose pool
    dispose_engine(engine)

    # Reconnect transparently works with new pool checkout
    with engine.connect() as conn:
        res = conn.execute(text("SELECT 2")).scalar()
        assert res == 2

    dispose_engine(engine)


def main() -> None:
    test_engine_creation_and_properties()
    test_engine_connect_and_execution()
    test_engine_begin_transaction_lifecycle()
    test_engine_connectionless_execution_removed()
    test_engine_dispose_cleanup()
    print(f"Contract verified successfully for sqlalchemy.Engine on {sa.__version__}")


if __name__ == "__main__":
    main()