SQLAlchemy stream 已调用 unique，转 scalars 后仍要求去重

异步查询使用集合 joinedload()，代码已经写了 result.unique().scalars()，读取时却仍报 The unique() method must be invoked on this Result。如果入口是 AsyncSession.stream()，要检查结果包装器有没有保留前一步的状态。

SQLAlchemy 2.1.5 中，AsyncResult.unique() 把去重状态放在异步结果自身，随后 scalars() 或 mappings() 却从底层同步结果建立新包装器，前面的状态便可能丢失。相同原因也会使 columns("b").scalars() 选回第一列，或使 mappings 返回本来已排除的列。

正式 2.1.6 已传递筛选状态

2.1.6 的正式实现从 AsyncResult 取得列元数据和去重状态，再以底层同步结果读取行。2.0 分支虽把对应修补列入 2.0.56，但当前尚无正式发行日期，不能把它当成已经可安装的修正版。升级应选择项目兼容且真实发行的版本，并从原来使用 stream 的入口验证；改成 AsyncSession.execute() 后正常，不足以证明流式路径已恢复，因为后者返回的是另一类结果。

暂时保留受影响依赖时，单个实体的去重场景可以将 unique() 放在最终的标量结果上：

result = await session.stream(stmt)
try:
    users = await result.scalars().unique().all()
finally:
    await result.close()

这里的 session 是已有 AsyncSession，stmt 仍是原来返回目标实体的查询，权限与加载选项不能删。顺序调整让去重状态直接设置在 AsyncScalarResult，而不是期待中间层替你传递。

列筛选不能一概靠调换顺序修好

上面的片段针对单实体 scalars。若原调用同时使用 columns() 或返回 mappings，应核对最终列集合和每列值，不能为了让异常消失删去列筛选。已有查询本来就需要指定某一列时，可在适用接口中明确选择，但不能把原来的多列行结构悄悄改成标量列表。

unique() 本身要记录已见的身份或值，也不保证任意大结果都保持固定内存。它与 ORM 的 yield_per 还有既有组合限制，本次修补没有取消那些限制。不能一边修状态丢失，一边无差别加上全量 .all() 来替换原来的流式业务。

回归应包含父实体有多个关联行、多个不同父实体及空结果，检查去重后的身份和关系内容；列筛选则同时验证键集合与标量值。只看异常消失或列表长度变小，仍可能漏掉选错列或丢行。