CodeSampleX

Exemple

greenlet 3.4.0: Handle SQLAlchemy 2.0 select joinedload collection deduplication requiring unique(), multi-column scalars() column truncation, and Row mapping access via _mapping

Échantillon vérifié pour pypi greenlet 3.4.0: Handle SQLAlchemy 2.0 select joinedload collection deduplication requiring unique(), multi-column scalars()…

sha256:4c2613bf98ad43fa0c7839a93eb9640e70f5bc939d32193e63ddcd7dd613d98f

Ce réseau offre une seule chose : un échantillon qui compile. Il l'a exécuté dans un bac à sable et conservé le reçu signé. Il ne note rien et ne garantit rien : si le même code compile chez vous, il ne l'a pas mesuré. Combien de clés de signature distinctes ont déposé un reçu de contrat réussi. Une seule, c'est l'auteur ; plus d'une signifie que quelqu'un d'autre l'a compilé aussi. Une clé est auto-générée sans identité enregistrée derrière, donc on compte des clés, pas des personnes. MIT-0

Preuves d'exécution

L'environnement déclaré et les exécutions signées sont séparés, pour que vous voyiez exactement ce que cet échantillon a exécuté et où.

Base de preuve
Contrat signé réussi
Reçus de vérification
3
Clés de signature qui l’ont compilé
3
Environnement déclaré python linux x64 python python pip

Environnements des exécutions de vérification

Environnement Contrat Étapes Exécution
python 3.12 · linux alpine/x64 · docker ed25519:d91480838ac982c9 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · python@1
2026-08-15
python 3.12 · linux alpine/x64 · docker ed25519:2175b912ea1c23b1 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · python@1
2026-08-18
python 3.12 · linux debian/x64 · docker ed25519:c1973797be207ac4 PASS compile:SKIPPED · contract:PASS · load:PASS · resolve:PASS
CONTAINER_RUN · python@1python:3.12-slim@sha256:09f7da3bc104…
2026-09-08

Cas

HOW
Objectif
Handle SQLAlchemy 2.0 select joinedload collection deduplication requiring unique(), multi-column scalars() column truncation, and Row mapping access via _mapping
Paquets
Symboles
  • sqlalchemy.select
  • sqlalchemy.orm.joinedload
  • sqlalchemy.orm.Session.scalars
  • sqlalchemy.orm.Session.execute
  • sqlalchemy.engine.Result.unique
  • sqlalchemy.engine.Result.mappings
  • sqlalchemy.engine.Row._mapping
Environnement
python
Créé
2026-08-15T14:42:30Z

Contrat

  1. assert calling scalars() with joinedload collection without unique() raises InvalidRequestError
  2. assert scalars() on multi-column select silently discards columns beyond index 0
  3. assert Row subscripting by string column name raises TypeError and requires ._mapping
  4. assert dict(row) raises TypeError and requires dict(row._mapping) or row._asdict()
  5. assert Result.mappings() returns RowMapping supporting key lookup and direct dict conversion

Fichiers

  • NOTES.md
  • csx.json
  • requirements.txt
  • src/__init__.py
  • src/queries.py
  • test/contract.py

Télécharger l’artefact source (tar.gz)

Code source

NOTES.md
# Notes: SQLAlchemy 2.0 Select Result Shapes & Joinedload Deduplication

## search_known_solution Result
HIT with sample `sha256:dc10ed4452bdb60c74525019b9ab846447e9b840a2857b5174c9db7250cd0196` ("Move a SQLAlchemy 1.4 query to the 2.0 select API and get entities back instead of Row tuples"). That sample focuses on wrapping single entity selects with `.scalars()`, `session.query().get()` deprecation warnings, `Engine.execute` removal, and bare string SQL exceptions. This sample targets a different trap suite: collection joined eager loading requiring explicit `.unique()`, `scalars()` silent column truncation on multi-column queries, and `Row` string indexing restrictions.

## What a Model Would Have Written Instead
```python
# 1. Eager loading collections without unique()
users = session.scalars(select(User).options(joinedload(User.addresses))).all()

# 2. Multi-column select expecting tuples or mapping
columns = session.scalars(select(User.id, User.name)).all()

# 3. Accessing columns by string key directly on Row
row = session.execute(select(User.id, User.name)).first()
name = row["name"]
row_dict = dict(row)
```

## How the Wrong Version Fails
- `joinedload` collection without `.unique()` fails **loudly** at runtime with `sqlalchemy.exc.InvalidRequestError: The unique() method must be invoked on this Result, as it contains results that include joined eager loads against collections`.
- `scalars()` on multi-column select fails **silently with a green build**, returning only the 0th column values (`[1, 2]`) while silently dropping all subsequent selected columns (`User.name`).
- `row["name"]` and `dict(row)` fail **loudly** at runtime with `TypeError: tuple indices must be integers or slices, not str` and `TypeError: cannot convert dictionary update sequence element #0 to a sequence` (requiring `row._mapping["name"]`, `row._asdict()`, or `.mappings()`).
csx.json
{"case":{"caseId":"case:sha256:e4c4e1b6231c30286d2497fdba01c3b3845f75ee940de218a44f263d6a37036d","constraints":{"runtime":"python"},"contract":["assert calling scalars() with joinedload collection without unique() raises InvalidRequestError","assert scalars() on multi-column select silently discards columns beyond index 0","assert Row subscripting by string column name raises TypeError and requires ._mapping","assert dict(row) raises TypeError and requires dict(row._mapping) or row._asdict()","assert Result.mappings() returns RowMapping supporting key lookup and direct dict conversion"],"goal":"Handle SQLAlchemy 2.0 select joinedload collection deduplication requiring unique(), multi-column scalars() column truncation, and Row mapping access via _mapping","kind":"HOW","packages":["pkg:pypi/greenlet@3.4.0","pkg:pypi/sqlalchemy@2.0.52","pkg:pypi/typing-extensions@4.15.0"],"schemaVersion":1,"symbols":["sqlalchemy.select","sqlalchemy.orm.joinedload","sqlalchemy.orm.Session.scalars","sqlalchemy.orm.Session.execute","sqlalchemy.engine.Result.unique","sqlalchemy.engine.Result.mappings","sqlalchemy.engine.Row._mapping"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/greenlet@3.4.0","pkg:pypi/sqlalchemy@2.0.52","pkg:pypi/typing-extensions@4.15.0"],"schemaVersion":1,"symbols":["sqlalchemy.select","sqlalchemy.orm.joinedload","sqlalchemy.orm.Session.scalars","sqlalchemy.orm.Session.execute","sqlalchemy.engine.Result.unique","sqlalchemy.engine.Result.mappings","sqlalchemy.engine.Row._mapping"],"verifierAdapter":"python@1"}
requirements.txt
sqlalchemy==2.0.52
typing-extensions==4.15.0
greenlet==3.4.0
src/__init__.py
"""SQLAlchemy 2.0 select and result shape samples."""
src/queries.py
import sqlalchemy as sa
from sqlalchemy import ForeignKey, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    joinedload,
    mapped_column,
    relationship,
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    addresses: Mapped[list["Address"]] = relationship(back_populates="user")


class Address(Base):
    __tablename__ = "addresses"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped[User] = relationship(back_populates="addresses")


def create_seeded_engine() -> sa.Engine:
    engine = sa.create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        u1 = User(
            id=1,
            name="ada",
            addresses=[
                Address(id=10, email="ada1@example.com"),
                Address(id=11, email="ada2@example.com"),
            ],
        )
        u2 = User(
            id=2,
            name="linus",
            addresses=[Address(id=20, email="linus@example.com")],
        )
        session.add_all([u1, u2])
        session.commit()
    return engine


def fetch_users_joined_eager(session: Session) -> list[User]:
    """2.0 select with joinedload requires explicit .unique() on the result."""
    return (
        session.scalars(select(User).options(joinedload(User.addresses)))
        .unique()
        .all()
    )


def fetch_user_columns_scalars(session: Session) -> list:
    """scalars() on multi-column SELECT silently discards all but the 0th column."""
    return session.scalars(select(User.id, User.name)).all()


def fetch_user_columns_rows(session: Session) -> list:
    """execute() returns Row tuples preserving all requested columns."""
    return session.execute(select(User.id, User.name)).all()


def fetch_user_mappings(session: Session) -> list:
    """mappings() returns RowMapping views supporting key lookup and dict() conversion."""
    return session.execute(select(User.id, User.name)).mappings().all()
test/contract.py
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import sqlalchemy as sa
from sqlalchemy import select
from sqlalchemy.exc import InvalidRequestError, MultipleResultsFound, NoResultFound
from sqlalchemy.orm import Session, joinedload

from src.queries import (
    Address,
    User,
    create_seeded_engine,
    fetch_user_columns_rows,
    fetch_user_columns_scalars,
    fetch_user_mappings,
    fetch_users_joined_eager,
)

engine = create_seeded_engine()

with Session(engine) as session:
    # 1. 2.0 joined eager loading on collections requires explicit .unique()
    # In 1.x Query, session.query(User).options(joinedload(User.addresses)).all() auto-deduplicated.
    # In 2.0 select, calling .all() without .unique() raises InvalidRequestError.
    try:
        session.scalars(select(User).options(joinedload(User.addresses))).all()
        raise AssertionError("Calling .all() on joinedload collection without .unique() should fail")
    except InvalidRequestError as exc:
        assert "unique()" in str(exc), exc

    # Explicit .unique() successfully returns deduplicated entities with eager collections
    users = fetch_users_joined_eager(session)
    assert len(users) == 2
    assert [a.email for a in users[0].addresses] == ["ada1@example.com", "ada2@example.com"]

    # 2. Multi-column select: scalars() silently extracts only the 0th column
    # A naive migration expecting tuples [(1, 'ada'), (2, 'linus')] gets [1, 2] with no warning.
    truncated = fetch_user_columns_scalars(session)
    assert truncated == [1, 2], truncated

    # execute() preserves all columns as Row objects
    rows = fetch_user_columns_rows(session)
    assert len(rows) == 2
    first_row = rows[0]
    assert type(first_row).__name__ == "Row"
    assert first_row[0] == 1 and first_row[1] == "ada"
    assert first_row.id == 1 and first_row.name == "ada"

    # 3. Row in 2.0 is a tuple sequence, NOT a mapping: string indexing raises TypeError
    try:
        _ = first_row["name"]
        raise AssertionError("Row['name'] should raise TypeError in 2.0")
    except TypeError as exc:
        assert "tuple indices must be integers or slices, not str" in str(exc), exc

    # String key access and dict conversion must go through ._mapping or ._asdict()
    assert first_row._mapping["name"] == "ada"
    assert first_row._asdict() == {"id": 1, "name": "ada"}
    assert dict(first_row._mapping) == {"id": 1, "name": "ada"}

    try:
        dict(first_row)
        raise AssertionError("dict(Row) directly should raise TypeError")
    except TypeError:
        pass

    # 4. Result.mappings() provides RowMapping where string indexing and dict() work directly
    mapping_rows = fetch_user_mappings(session)
    first_m = mapping_rows[0]
    assert type(first_m).__name__ == "RowMapping"
    assert first_m["name"] == "ada"
    assert dict(first_m) == {"id": 1, "name": "ada"}

    # 5. Cardinality rules on 2.0 results
    # scalar_one() / scalars().one() raises MultipleResultsFound when > 1 row
    try:
        session.scalars(select(User)).one()
        raise AssertionError("scalars().one() on multi-row result should raise MultipleResultsFound")
    except MultipleResultsFound:
        pass

    # scalars().one() raises NoResultFound when 0 rows
    try:
        session.scalars(select(User).where(User.id == 999)).one()
        raise AssertionError("scalars().one() on empty result should raise NoResultFound")
    except NoResultFound:
        pass

    # first() and one_or_none() return None on empty result
    assert session.scalars(select(User).where(User.id == 999)).first() is None
    assert session.scalars(select(User).where(User.id == 999)).one_or_none() is None

print("contract ok:", sa.__version__)

Seeder d'origine

anonymous