A minimal example how pytsql changes the autocommit behaviour of engines

pytsql_break_minimal_example.py

import random
import string
import pytest

import pytsql

import sqlalchemy
from sqlalchemy.pool import NullPool


TEMPDB_CONN_STR = "mssql+pyodbc://sa:QuantCo123@localhost:1433/tempdb?driver=libtdsodbc.so"


def show_bug(
    tempdb_connection_string: str,
    break_with_root_cause: bool,
    fix_by_disabling_pooling: bool,
    db: str = "tempdb",
):
    eng = sqlalchemy.create_engine(
        tempdb_connection_string,
        connect_args={"autocommit": True},
        echo=False,
        **({"poolclass": NullPool} if fix_by_disabling_pooling else {}),
    )

    source_table = random_name()
    table_1 = random_name()
    table_2 = random_name()

    # print("Source table:", source_table)
    # print("Table 1:", table_1)
    # print("Table 2:", table_2)

    with eng.connect() as conn:
        conn.execute(f"CREATE TABLE {db}.dbo.{source_table} (ID INT)")
        conn.execute(f"INSERT INTO {db}.dbo.{source_table} VALUES (123)")

    with eng.connect() as conn:
        conn.execute(f"SELECT * INTO {db}.dbo.{table_1} FROM {db}.dbo.{source_table}")
    with eng.connect() as conn:
        print("Found in table:", conn.execute(f"SELECT * FROM {db}.dbo.{table_1}").all())

    if not break_with_root_cause:
        pytsql.executes("SELECT 12", eng)
    else:
        eng.connect().execution_options(isolation_level="AUTOCOMMIT")

    with eng.connect() as conn:
        # Won't be committed if no fixes are applied.
        conn.execute(f"SELECT * INTO {db}.dbo.{table_2} FROM {db}.dbo.{source_table}")
    with eng.connect() as conn:
        # Fails with "invalid object name" if no fixes are applied because the above didn't get committed.
        print("Found in table:", conn.execute(f"SELECT * FROM {db}.dbo.{table_2}").all())


def random_name() -> str:
    return "".join(random.choices(string.ascii_letters, k=10))


print("Should work")
for break_with_root_cause in [False, True]:
    show_bug(
        tempdb_connection_string=TEMPDB_CONN_STR,
        break_with_root_cause=break_with_root_cause,
        fix_by_disabling_pooling=True,  # Disabling pooling fixes it because connections are not reused.
    )

print("Should NOT work")
for break_with_root_cause in [False, True]:
    with pytest.raises(sqlalchemy.exc.ProgrammingError, match="Invalid object name"):
        show_bug(
            tempdb_connection_string=TEMPDB_CONN_STR,
            break_with_root_cause=break_with_root_cause,
            fix_by_disabling_pooling=False,
        )
    print("Told ya")