SQLModel tables queried with SQLAlchemy Core select on sqlite: working versions (tested 2026-10-16)

sqlmodel==0.0.48
sqlalchemy==2.1.4
pytest==9.1.1

from typing import Optional

from sqlmodel import Field, SQLModel, create_engine


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    team: str = "avengers"
    power: int = 0


def create_db_engine():
    engine = create_engine("sqlite://")
    SQLModel.metadata.create_all(engine)
    return engine


def seed(engine) -> None:
    from sqlmodel import Session

    with Session(engine) as session:
        session.add_all(
            [
                Hero(name="iron", team="avengers", power=7),
                Hero(name="thor", team="avengers", power=9),
                Hero(name="loki", team="villains", power=6),
            ]
        )
        session.commit()

from sqlalchemy import func, insert
from sqlalchemy import select as core_select

from models import Hero


def names_on_team(session, team: str) -> list[str]:
    # Core select of one column still yields Row tuples, so unwrap with scalars().
    stmt = core_select(Hero.name).where(Hero.team == team).order_by(Hero.name)
    return list(session.exec(stmt).scalars())


def team_sizes(session) -> list[tuple[str, int]]:
    # A multi-column Core select returns Rows accessible by attribute or index.
    stmt = (
        core_select(Hero.team, func.count(Hero.id).label("n"))
        .group_by(Hero.team)
        .order_by(Hero.team)
    )
    return [(row.team, row.n) for row in session.exec(stmt)]


def strongest(session) -> Hero | None:
    # A Core select of the entity wraps each model in a Row tuple; unpack it.
    stmt = core_select(Hero).order_by(Hero.power.desc())
    row = session.exec(stmt).first()
    return row[0] if row else None


def total_heroes(session) -> int:
    # Core count() comes back as a one-column Row; SQLModel's select() does not.
    return session.exec(core_select(func.count()).select_from(Hero)).one()[0]


def recruit(session, name: str, team: str, power: int) -> int:
    # Core insert with RETURNING: read the value before committing on sqlite.
    result = session.execute(
        insert(Hero).values(name=name, team=team, power=power).returning(Hero.id)
    )
    new_id = result.scalar_one()
    session.commit()
    return new_id

import pytest
from sqlalchemy import Row, func
from sqlalchemy import select as core_select
from sqlmodel import Session, select

from models import Hero, create_db_engine, seed
from queries import names_on_team, recruit, strongest, team_sizes, total_heroes


@pytest.fixture()
def session():
    engine = create_db_engine()
    seed(engine)
    with Session(engine) as s:
        yield s


def test_core_select_does_not_scalarize_one_column(session):
    stmt = core_select(Hero.name).order_by(Hero.name)
    rows = session.exec(stmt).all()
    assert all(isinstance(row, Row) for row in rows)
    assert rows[0][0] == "iron"
    assert list(session.exec(stmt).scalars()) == ["iron", "loki", "thor"]


def test_sqlmodel_select_scalarizes_one_column(session):
    stmt = select(Hero.name).order_by(Hero.name)
    assert list(session.exec(stmt)) == ["iron", "loki", "thor"]


def test_entity_core_select_yields_row_tuples(session):
    core_rows = session.exec(core_select(Hero)).all()
    assert all(isinstance(row, Row) for row in core_rows)
    assert core_rows[0][0].name == "iron"

    models = session.exec(select(Hero)).all()
    assert all(isinstance(row, Hero) for row in models)


def test_group_by_rows_read_by_attribute_or_index(session):
    assert team_sizes(session) == [("avengers", 2), ("villains", 1)]


def test_count_shape_differs_between_select_helpers(session):
    assert total_heroes(session) == 3
    assert session.exec(select(func.count()).select_from(Hero)).one() == 3


def test_core_insert_returning(session):
    new_id = recruit(session, "odin", "avengers", 10)
    best = strongest(session)
    assert best.name == "odin"
    assert best.id == new_id

[pytest]
pythonpath = .