sqlalchemy 2.0.51: create_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.create_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.create_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:4197ab723c6ce88339f2c1fe237bea2d7ff86dbd6cff1662d3fa5e867120d71d","contract":["create_engine returns an Engine instance configured with the specified URL dialect, driver, and execution options.","create_engine connect() provides connection context management and executes SQL statements.","create_engine begin() provides transactional context management with automatic commit on normal exit and rollback on exception.","create_engine supports connection pool configuration and engine.dispose() to close pooled connections.","create_engine raises ArgumentError when provided an invalid database URL string."],"goal":"verify sqlalchemy.create_engine in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.create_engine"]},"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.create_engine"],"verifierAdapter":"python@1"}

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

"""SQLAlchemy 2.0 create_engine usage and verification sample."""

from typing import List, Tuple
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine


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


def setup_schema(engine: Engine) -> None:
    """Create sample tables using an engine transaction."""
    with engine.begin() as conn:
        conn.execute(
            text(
                "CREATE TABLE IF NOT EXISTS users ("
                "  id INTEGER PRIMARY KEY AUTOINCREMENT,"
                "  username TEXT NOT NULL,"
                "  email TEXT NOT NULL"
                ")"
            )
        )


def insert_user(engine: Engine, username: str, email: str) -> int:
    """Insert a user and return the inserted row id."""
    with engine.begin() as conn:
        result = conn.execute(
            text("INSERT INTO users (username, email) VALUES (:username, :email)"),
            {"username": username, "email": email},
        )
        return result.lastrowid or 0


def fetch_all_users(engine: Engine) -> List[Tuple[int, str, str]]:
    """Query and return all user records."""
    with engine.connect() as conn:
        result = conn.execute(text("SELECT id, username, email FROM users ORDER BY id"))
        return [(row.id, row.username, row.email) for row in result]


if __name__ == "__main__":
    engine = initialize_engine("sqlite:///:memory:", echo=False)
    setup_schema(engine)
    user_id = insert_user(engine, "alice", "alice@example.com")
    users = fetch_all_users(engine)
    print(f"Inserted user id: {user_id}, total users: {len(users)}")
    engine.dispose()

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

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

import os
import sys

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

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine, Connection
from sqlalchemy.exc import ArgumentError
from sample import (
    initialize_engine,
    setup_schema,
    insert_user,
    fetch_all_users,
)


def test_engine_creation_and_properties() -> None:
    # 1. create_engine returns an Engine instance configured with the specified URL dialect, driver, and execution options.
    engine = initialize_engine("sqlite:///:memory:", echo=False)
    assert isinstance(engine, Engine), f"Expected Engine instance, got {type(engine)}"
    assert engine.name == "sqlite"
    assert engine.dialect.name == "sqlite"
    assert str(engine.url) == "sqlite:///:memory:"
    assert engine.echo is False
    engine.dispose()


def test_engine_connect() -> None:
    # 2. create_engine connect() provides connection context management and executes SQL statements.
    engine = create_engine("sqlite:///:memory:")
    with engine.connect() as conn:
        assert isinstance(conn, Connection)
        result = conn.execute(text("SELECT 1 + 1 AS sum_val"))
        row = result.fetchone()
        assert row is not None
        assert row[0] == 2
        assert row.sum_val == 2
    engine.dispose()


def test_engine_begin_transactions() -> None:
    # 3. create_engine begin() provides transactional context management with automatic commit on normal exit and rollback on exception.
    engine = create_engine("sqlite:///:memory:")
    setup_schema(engine)

    # Insert user in a committed transaction
    user_id = insert_user(engine, "alice", "alice@example.com")
    assert user_id > 0

    users = fetch_all_users(engine)
    assert len(users) == 1
    assert users[0][1] == "alice"

    # Verify rollback on exception in begin() block
    try:
        with engine.begin() as conn:
            conn.execute(
                text("INSERT INTO users (username, email) VALUES (:username, :email)"),
                {"username": "bob", "email": "bob@example.com"},
            )
            raise RuntimeError("Intentional failure to trigger rollback")
    except RuntimeError:
        pass

    users_after_rollback = fetch_all_users(engine)
    assert len(users_after_rollback) == 1
    engine.dispose()


def test_engine_pool_and_dispose() -> None:
    # 4. create_engine supports connection pool configuration and engine.dispose() to close pooled connections.
    pooled_engine = create_engine("sqlite:///:memory:", pool_pre_ping=True)
    with pooled_engine.connect() as conn:
        result = conn.execute(text("SELECT 42"))
        assert result.scalar() == 42
    pooled_engine.dispose()


def test_engine_invalid_url() -> None:
    # 5. create_engine raises ArgumentError when provided an invalid database URL string.
    try:
        create_engine("invalid_url_without_scheme")
        raise AssertionError("Expected ArgumentError for invalid database URL")
    except ArgumentError:
        pass


def main() -> None:
    test_engine_creation_and_properties()
    test_engine_connect()
    test_engine_begin_transactions()
    test_engine_pool_and_dispose()
    test_engine_invalid_url()
    print("CONTRACT PASS: sqlalchemy.create_engine verified offline")


if __name__ == "__main__":
    main()