pandas to_sql and read_sql with a SQLAlchemy 2 engine on sqlite, with dtype handling: working versions (tested 2026-10-15)

pandas==3.0.6
sqlalchemy==2.1.4
pytest==9.1.1
numpy==2.3.4
python-dateutil==2.9.0.post0
six==1.17.0
typing-extensions==4.15.0
iniconfig==2.1.0
pluggy==1.6.0
pygments==2.19.2

import pandas as pd
import sqlalchemy as sa
from sqlalchemy.pool import StaticPool

def make_engine(url: str = "sqlite://") -> sa.Engine:
    # StaticPool keeps one shared connection, so every session sees the same in-memory DB
    return sa.create_engine(url, poolclass=StaticPool)

def save_frame(df: pd.DataFrame, engine: sa.Engine, table: str,
               dtype: dict | None = None, if_exists: str = "replace") -> int | None:
    # index=False: pandas defaults to index=True and would add an "index" column
    # engine.begin() commits on exit; engine.connect() alone would roll back
    with engine.begin() as conn:
        return df.to_sql(table, conn, index=False, dtype=dtype, if_exists=if_exists)

def load_frame(engine: sa.Engine, query: str, **kwargs) -> pd.DataFrame:
    # SQLAlchemy 2 requires text() for raw SQL strings; bind params via params=, never f-strings
    with engine.connect() as conn:
        return pd.read_sql(sa.text(query), conn, **kwargs)

import pandas as pd
import pytest
import sqlalchemy as sa
from db import make_engine, save_frame, load_frame

def orders():
    return pd.DataFrame({
        "sku": ["A1", "B2", "C3"],
        "qty": pd.array([1, None, 3], dtype="Int64"),
        "price": [9.5, 4.0, 12.25],
        "ts": pd.to_datetime(["2024-01-01 10:00", "2024-01-02 11:30", "2024-01-03 09:15"]),
    })

@pytest.fixture
def engine():
    return make_engine()

def test_no_index_column_written(engine):
    save_frame(orders(), engine, "orders")
    cols = [c["name"] for c in sa.inspect(engine).get_columns("orders")]
    assert cols == ["sku", "qty", "price", "ts"]

def test_dtype_controls_column_types(engine):
    save_frame(orders(), engine, "orders", dtype={"sku": sa.String(10), "qty": sa.Integer})
    types = {c["name"]: str(c["type"]) for c in sa.inspect(engine).get_columns("orders")}
    assert types["sku"] == "VARCHAR(10)" and types["qty"] == "INTEGER"

def test_nullable_int_roundtrip(engine):
    save_frame(orders(), engine, "orders")
    # Without dtype_backend="numpy_nullable", NULLs come back as float NaN, not Int64 <NA>
    back = load_frame(engine, "SELECT * FROM orders ORDER BY sku", dtype_backend="numpy_nullable")
    assert str(back["qty"].dtype) == "Int64"
    assert back["qty"].isna().tolist() == [False, True, False]

def test_datetimes_are_text_until_parse_dates(engine):
    save_frame(orders(), engine, "orders")
    # sqlite has no datetime type, so pandas stores timestamps as text
    raw = load_frame(engine, "SELECT ts FROM orders")
    assert isinstance(raw["ts"].iloc[0], str)
    parsed = load_frame(engine, "SELECT ts FROM orders", parse_dates=["ts"])
    assert pd.api.types.is_datetime64_any_dtype(parsed["ts"])

def test_bound_params_instead_of_string_formatting(engine):
    save_frame(orders(), engine, "orders")
    out = load_frame(engine, "SELECT sku FROM orders WHERE sku = :s", params={"s": "B2"})
    assert out["sku"].tolist() == ["B2"]

def test_if_exists_fail_raises(engine):
    save_frame(orders(), engine, "orders")
    with pytest.raises(ValueError):
        save_frame(orders(), engine, "orders", if_exists="fail")

[pytest]
pythonpath = .