sqlalchemy 2.0.51: pool.SingletonThreadPool

# sqlalchemy SingletonThreadPool API

This sample pins `sqlalchemy` 2.0.51 and verifies `sqlalchemy.pool.SingletonThreadPool` for thread-local connection pooling.

The contract verifies:
- SingletonThreadPool initializes with a creator callable and pool_size configuration.
- SingletonThreadPool maintains and reuses a single connection per thread across multiple connect calls.
- SingletonThreadPool allocates distinct connections for different threads.
- SingletonThreadPool.dispose safely closes active pooled connections and resets thread-local state.
- SingletonThreadPool.recreate creates a new pool preserving configuration parameters.
- create_engine accepts SingletonThreadPool as poolclass and executes queries successfully.

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.pool.SingletonThreadPool 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.pool.SingletonThreadPool

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:219c2be43030c7a1f9d4fe97321f4f0d455678bd18087e27d899e944a8a9de8e","contract":["SingletonThreadPool initializes with a creator callable and pool_size configuration.","SingletonThreadPool maintains and reuses a single connection per thread across multiple connect calls.","SingletonThreadPool allocates distinct connections for different threads.","SingletonThreadPool.dispose safely closes active pooled connections and resets thread-local state.","SingletonThreadPool.recreate creates a new pool preserving configuration parameters.","create_engine accepts SingletonThreadPool as poolclass and executes queries successfully."],"goal":"verify sqlalchemy.pool.SingletonThreadPool in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.pool.SingletonThreadPool"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","executionContext":"python","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.pool.SingletonThreadPool"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51

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

"""Contract verification tests for sqlalchemy.pool.SingletonThreadPool."""

import sqlite3
import threading
from sqlalchemy import create_engine, text
from sqlalchemy.pool import SingletonThreadPool
from sqlalchemy.pool.base import _ConnectionFairy


def test_singleton_thread_pool_init():
    """Verify SingletonThreadPool initializes with a creator callable and pool_size configuration."""
    creator_called = 0

    def make_conn():
        nonlocal creator_called
        creator_called += 1
        return sqlite3.connect(":memory:", check_same_thread=False)

    pool = SingletonThreadPool(make_conn, pool_size=5)
    assert isinstance(pool, SingletonThreadPool), f"Expected SingletonThreadPool instance, got {type(pool)}"
    assert pool.size == 5
    assert creator_called == 0
    pool.dispose()


def test_singleton_thread_pool_reuse_in_thread():
    """Verify SingletonThreadPool maintains and reuses a single connection per thread across multiple connect calls."""
    creator_called = 0

    def make_conn():
        nonlocal creator_called
        creator_called += 1
        return sqlite3.connect(":memory:", check_same_thread=False)

    pool = SingletonThreadPool(make_conn, pool_size=5)

    conn1 = pool.connect()
    assert isinstance(conn1, _ConnectionFairy)
    raw_conn1 = conn1.dbapi_connection
    assert creator_called == 1
    conn1.close()

    conn2 = pool.connect()
    raw_conn2 = conn2.dbapi_connection
    assert creator_called == 1, "Creator should not be invoked again in the same thread"
    assert raw_conn1 is raw_conn2, "Expected identical DBAPI connection to be reused in the same thread"
    conn2.close()

    pool.dispose()


def test_singleton_thread_pool_thread_isolation():
    """Verify SingletonThreadPool allocates distinct connections for different threads."""
    creator_called = 0

    def make_conn():
        nonlocal creator_called
        creator_called += 1
        return sqlite3.connect(":memory:", check_same_thread=False)

    pool = SingletonThreadPool(make_conn, pool_size=5)

    main_conn = pool.connect()
    main_raw = main_conn.dbapi_connection
    main_conn.close()

    thread_conns = []

    def worker():
        c = pool.connect()
        thread_conns.append(c.dbapi_connection)
        c.close()

    t1 = threading.Thread(target=worker)
    t2 = threading.Thread(target=worker)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    assert creator_called == 3
    assert len(thread_conns) == 2
    assert thread_conns[0] is not main_raw
    assert thread_conns[1] is not main_raw
    assert thread_conns[0] is not thread_conns[1], "Distinct worker threads must have distinct connections"

    pool.dispose()


def test_singleton_thread_pool_dispose():
    """Verify SingletonThreadPool.dispose safely closes active pooled connections and resets thread-local state."""
    creator_called = 0

    def make_conn():
        nonlocal creator_called
        creator_called += 1
        return sqlite3.connect(":memory:", check_same_thread=False)

    pool = SingletonThreadPool(make_conn, pool_size=5)

    conn1 = pool.connect()
    raw_conn1 = conn1.dbapi_connection
    conn1.close()

    pool.dispose()

    conn2 = pool.connect()
    raw_conn2 = conn2.dbapi_connection
    assert creator_called == 2, "Creator must be called again after pool.dispose()"
    assert raw_conn2 is not raw_conn1, "New DBAPI connection must be instantiated after pool.dispose()"
    conn2.close()

    pool.dispose()


def test_singleton_thread_pool_recreate():
    """Verify SingletonThreadPool.recreate creates a new pool preserving configuration parameters."""
    creator_called = 0

    def make_conn():
        nonlocal creator_called
        creator_called += 1
        return sqlite3.connect(":memory:", check_same_thread=False)

    pool = SingletonThreadPool(make_conn, pool_size=4)
    recreated = pool.recreate()

    assert isinstance(recreated, SingletonThreadPool)
    assert recreated.size == 4
    assert recreated is not pool

    recreated.dispose()
    pool.dispose()


def test_engine_integration():
    """Verify create_engine accepts SingletonThreadPool as poolclass and executes queries successfully."""
    engine = create_engine(
        "sqlite:///:memory:",
        poolclass=SingletonThreadPool,
        pool_size=3,
        connect_args={"check_same_thread": False},
    )
    assert isinstance(engine.pool, SingletonThreadPool)
    assert engine.pool.size == 3

    with engine.connect() as conn:
        result = conn.scalar(text("SELECT 42"))
        assert result == 42

    engine.dispose()


if __name__ == "__main__":
    test_singleton_thread_pool_init()
    test_singleton_thread_pool_reuse_in_thread()
    test_singleton_thread_pool_thread_isolation()
    test_singleton_thread_pool_dispose()
    test_singleton_thread_pool_recreate()
    test_engine_integration()
    print("ALL CONTRACT TESTS PASSED")