SQLAlchemy中如何为ORM实体分配列名以实现命名属性访问？

在SQLAlchemy 1.4中，我编写了如下查询，选择ORM实体models.Person并附加prev/next计算列：

@pytest.mark.wip
@pytest.mark.asyncio
async def test_sqlalchemy(people: AsyncSession) -> None:
    session = people

    query = (
        select(
            models.Person,
            Bundle(
                "navigation",
                sa.func.lag(models.Person.cursor)
                .over(order_by=models.Person.cursor)
                .label("prev"),
                sa.func.lead(models.Person.cursor)
                .over(order_by=models.Person.cursor)
                .label("next"),
            ),
        )
        .where(models.Person.id > "26744d86-1918-4f67-92a0-e7ca12043721")
        .order_by(models.Person.id.asc())
    )

    iterator = await session.execute(query)
    rows = iterator.all()

    print(f"person = {rows[0][0]}")
    print(f"next={rows[0].navigation.next} prev={rows[0].navigation.prev}")

执行后，计算列可通过命名属性（如rows[0].navigation.next）访问，但Person实体只能通过索引（如rows[0][0].surname）访问。尝试用Bundle("person", models.Person)包裹实体时，出现错误：AttributeError: 'AnnotatedTable' object has no attribute '_label'。请问如何实现rows[0].person.surname这样的命名访问？

解决方案：客户端结果转换（简单直接） #

由于SQLAlchemy的Bundle仅支持包装列元素（ColumnElement），无法直接包裹ORM实体，最简便的方法是在查询结果返回后，将每行数据转换为命名元组（namedtuple），从而实现实体的命名属性访问：

from collections import namedtuple

@pytest.mark.wip
@pytest.mark.asyncio
async def test_sqlalchemy(people: AsyncSession) -> None:
    session = people

    query = (
        select(
            models.Person,
            Bundle(
                "navigation",
                sa.func.lag(models.Person.cursor)
                .over(order_by=models.Person.cursor)
                .label("prev"),
                sa.func.lead(models.Person.cursor)
                .over(order_by=models.Person.cursor)
                .label("next"),
            ),
        )
        .where(models.Person.id > "26744d86-1918-4f67-92a0-e7ca12043721")
        .order_by(models.Person.id.asc())
    )

    iterator = await session.execute(query)
    # 定义命名元组，对应查询中的两个返回项
    PersonNavigationRow = namedtuple('PersonNavigationRow', ['person', 'navigation'])
    # 将结果转换为命名元组列表
    rows = [PersonNavigationRow(*row) for row in iterator.all()]

    # 现在可以通过命名属性访问
    print(f"person = {rows[0].person}")
    print(f"next={rows[0].navigation.next} prev={rows[0].navigation.prev}")
    print(f"person surname: {rows[0].person.surname}")

原理说明

• 命名元组会将查询返回的元组（Person实体 + navigation Bundle）映射为带有指定名称的属性，完全符合你需要的rows[0].person.surname访问方式。
• 这种方式不需要修改SQL查询结构，仅在结果处理层做转换，避免了ORM实体与Bundle兼容的问题。

备选方案：使用别名实体（SQL层面） #

如果你希望在SQL层面实现命名关联，可以使用sqlalchemy.orm.aliased为Person实体创建别名，再通过结果映射访问：

import sqlalchemy as sa
from sqlalchemy.orm import aliased

@pytest.mark.wip
@pytest.mark.asyncio
async def test_sqlalchemy(people: AsyncSession) -> None:
    session = people
    # 创建Person实体的别名
    person_alias = aliased(models.Person, name='person')

    query = (
        select(
            person_alias,
            Bundle(
                "navigation",
                sa.func.lag(person_alias.cursor)
                .over(order_by=person_alias.cursor)
                .label("prev"),
                sa.func.lead(person_alias.cursor)
                .over(order_by=person_alias.cursor)
                .label("next"),
            ),
        )
        .where(person_alias.id > "26744d86-1918-4f67-92a0-e7ca12043721")
        .order_by(person_alias.id.asc())
    )

    iterator = await session.execute(query)
    # 使用mappings()获取字典映射结果
    mappings = iterator.mappings().all()
    # 通过别名对象访问实体
    print(f"person surname: {mappings[0][person_alias].surname}")
    print(f"next={mappings[0]['navigation'].next}")

这种方式需要通过别名对象作为键访问实体，虽然不如命名元组直观，但属于SQL层面的别名设置。

内容的提问来源于stack exchange，提问作者anon_dcs3spp