Result

官方文档: https://docs.sqlalchemy.org.cn/en/20/core/connections.html#sqlalchemy.engine.Result

下面给你一版最全、最系统、SQLAlchemy 2.0 Result / ScalarResult / Row / RowMapping 的方法详解

覆盖同步/异步通用 API，包含常用、少用、底层细节与性能特性，适合你现在大量使用 SQLAlchemy ORM + Core 的场景。

⚠️ SQLAlchemy 2.0 的结果系统与 1.4 完全重写：

不再返回 ResultProxy
结果由 Result / ScalarResult / ChunkedIteratorResult 组成
行类型为 Row，索引结构在 Row._mapping 中

🚀 一、SQLAlchemy Result 的类型体系

SQLAlchemy 2.0 中，有 3 种主要结果类型：

类型	出现场景	说明
Result	执行 session.execute(select…)	行为对象结果，包含 Row
ScalarResult	执行 session.scalars(select…) 或 result.scalars()	单值结果（单个或单列），如 count()、select(User.id)
CursorResult	执行 Core 原生 SQL (connection.execute(text))	直接游标结果

并且 Result 内部行对象是：

类型	描述
Row	类似 tuple，可通过 index、attr、key 访问
RowMapping	row._mapping，真正的 dict-like 结构

☘️ 二、Result 方法大全

⚠️ 注意：Result 和 ScalarResult 只能调用一次各自的方法，然后就会被关闭。要调用其它方法，需要再次执行SQL。

方法列表：

dir(sqlalchemy.Result)

# ['__annotations__', '__class__', '__class_getitem__', '__delattr__', '__dict__', '__dir__', '__doc__', '__enter__', '__eq__', '__exit__', '__format__', '__ge__', '__getattribute__', '__getstate__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__next__', '__orig_bases__', '__parameters__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__slots__', '__str__', '__subclasshook__', '_allrows', '_assert_no_memoizations', '_attributes', '_column_slices', '_fetchall_impl', '_fetchiter_impl', '_fetchmany_impl', '_fetchone_impl', '_generate', '_generate_rows', '_getter', '_is_cursor', '_iter_impl', '_iterator_getter', '_manyrow_getter', '_memoized_keys', '_metadata', '_next_impl', '_onerow_getter', '_only_one_row', '_post_creational_filter', '_raw_all_rows', '_raw_row_iterator', '_real_result', '_reset_memoizations', '_row_getter', '_row_logging_fn', '_set_memoized_attribute', '_soft_close', '_soft_closed', '_source_supports_scalars', '_tuple_getter', '_unique_filter_state', '_unique_strategy', '_yield_per', 'all', 'close', 'closed', 'columns', 'fetchall', 'fetchmany', 'fetchone', 'first', 'freeze', 'keys', 'mappings', 'memoized_attribute', 'memoized_instancemethod', 'merge', 'one', 'one_or_none', 'partitions', 'scalar', 'scalar_one', 'scalar_one_or_none', 'scalars', 't', 'tuples', 'unique', 'yield_per']

方法说明：

方法	说明	备注
all	以序列形式返回所有行	常用 (将结果转换为 list 类型的数据)
close	关闭此 Result
closed	如果此 Result 报告 .closed，则返回 True
columns	建立每行应返回的列
fetchall	all 方法的同义词
fetchmany	提取多行
fetchone	提取一行
first	提取第一行，如果没有行则返回 None
freeze	返回一个可调用对象
keys	返回一个包含所有字段名的序列
mappings	返回一个 MappingResult	常用 (将结果转换为 `list[dict]` 类型的数据)
memoized_attribute
memoized_instancemethod
merge	将此 Result 与其他兼容的结果对象合并
one	返回恰好一行，否则引发异常 (必须返回 1 行)
one_or_none	最多返回一个结果，否则引发异常 (0 或 1 行)
partitions	迭代处理给定大小的行子列表
scalar	获取第一行的第一列
scalar_one	返回恰好一个标量结果，否则引发异常
scalar_one_or_none	返回恰好一个标量结果，或者 None
scalars	返回一个 ScalarResult
t	tuples 方法的同义词
tuples	将 “类型化元组” 类型过滤器应用于返回的行	常用 (将结果转换为 tuple 类型的数据)
unique	对此 Result 返回的对象应用唯一性过滤
yield_per	配置行提取策略，一次提取 num 行

📒 多数情况下，对象都有方法。如果返回的是对象，则调用对象的方法，不要直接使用 list 或 tuple 转换数据！

返回对象的方法	获取数据
columns	调用方法
keys	list 或 tuple
mappings	调用方法
partitions	list 或 tuple
scalars	调用方法
tuples	调用方法
unique	调用方法

2.1 all()

返回所有行（Row 对象）列表。

rows: list[Row] = result.all()

2.2 first()

返回第一行（或 None）。

row = result.first()

2.3 one()

必须返回且只能返回一行，否则异常。

row = result.one()

错误：

返回 0 行 -> NoResultFound
返回多行 -> MultipleResultsFound

2.4 one_or_none()

0 行 -> None
1 行 -> 返回

row = result.one_or_none()

2.5 fetchone(), fetchmany(), fetchall()

等同于 DB-API，游标式取数，只对 CursorResult 有意义。

row = result.fetchone()
rows = result.fetchall()

ORM Result 多用 .all() 而不是 .fetchall()。

2.6 scalar()

返回第一行第一列的标量值。

val = result.scalar()

2.7 scalar_one()

必须返回且只能返回一个标量。

val = result.scalar_one()

2.8 scalar_one_or_none()

0 行 -> None
1 行 -> 标量

1 行 -> 异常

2.9 scalars()

将 Result 转换为 ScalarResult。

常用：

ids = result.scalars().all()

2.10 mappings()

返回字典行（RowMapping）

result = session.execute(select(User.id, User.username))
rows = result.mappings().all()

# rows = [{"id":1,"username":"a"}, ...]

极实用，避免 Row -> dict 手动处理。

🌟🌟 优雅的转换为 list[dict] 类型的数据 🌟🌟

dicts = [dict(row) for row in result.mappings().all()]

2.11 keys()

返回字段名列表：

cols = result.keys()

2.12 partitions(size)

按块迭代结果，提高大数据集性能。

for chunk in result.partitions(1000):
    ...

2.13 tuples()

强制结果每行转 tuple。

tuples = result.tuples().all()

2.14 is_singleton / returns_rows

属性：

是否返回行
是否只有一个列
是否已关闭

示例：

result.returns_rows  # True / False
result.returns_columns

2.15 rowcount

对 UPDATE/DELETE 有用：

affected = result.rowcount

2.16 close()

手动关闭游标。

🎯 三、ScalarResult 方法（scalars() 的结果）

ScalarResult 是一列的结果

方法列表：

dir(sqlalchemy.ScalarResult)

# ['__annotations__', '__class__', '__class_getitem__', '__delattr__', '__dict__', '__dir__', '__doc__', '__enter__', '__eq__', '__exit__', '__format__', '__ge__', '__getattribute__', '__getstate__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__next__', '__orig_bases__', '__parameters__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__slots__', '__str__', '__subclasshook__', '_allrows', '_assert_no_memoizations', '_attributes', '_column_slices', '_fetchall_impl', '_fetchiter_impl', '_fetchmany_impl', '_fetchone_impl', '_generate', '_generate_rows', '_is_cursor', '_iter_impl', '_iterator_getter', '_manyrow_getter', '_memoized_keys', '_metadata', '_next_impl', '_onerow_getter', '_only_one_row', '_post_creational_filter', '_raw_all_rows', '_real_result', '_reset_memoizations', '_row_getter', '_set_memoized_attribute', '_soft_close', '_soft_closed', '_unique_filter_state', '_unique_strategy', 'all', 'close', 'closed', 'fetchall', 'fetchmany', 'first', 'memoized_attribute', 'memoized_instancemethod', 'one', 'one_or_none', 'partitions', 'unique', 'yield_per']

方法说明：

方法	说明	适用范围
all	返回所有标量	单列多行
close	关闭此 FilterResult
closed	如果底层 Result 报告已关闭，则返回 True
fetchall	all 方法的同义词	单列多行
fetchmany	获取多个对象
first	获取第一个对象，如果没有对象则返回 None	单列多行
memoized_attribute
memoized_instancemethod
one	最多返回一个对象，否则引发异常 (必须返回 1 个)	单列单行 (单个)
one_or_none	最多返回一个对象，否则引发异常 (0 或 1 个)	单列单行 (单个)
partitions	迭代给定大小的元素子列表
unique	唯一性过滤 (去重)	单列多行
yield_per	配置行提取策略，一次提取 num 行

示例：

result = session.scalars(select(User.username))
names = result.all()

📒 笔记: ScalarResult 既可以是 单列单行 (比如 count()) ，也可以是 单列多行 (比如 select id from table ...)。

🎯 四、Row 和 RowMapping 方法

Row（行对象）

可通过三种方式访问：

row[0]
row["username"]
row.username

RowMapping（字典视图）

🚀 五、你真正需要的“常用组合”

5.1 返回 ORM 对象列表

items = session.execute(select(User)).scalars().all()

5.2 返回 dict 列表（最实用）

rows = session.execute(select(User.id, User.username)).mappings().all()

5.3 返回 tuple

tuples = session.execute(select(User.id, User.username)).tuples().all()

5.4 update/delete 返回影响行数

result = session.execute(update(User).values(name="xx"))
print(result.rowcount)

🎁 六、总结大表

功能	推荐方法	说明
返回 ORM 对象	scalars().all()	最常用
返回 dict	mappings().all()	强烈推荐
返回 tuple	tuples().all()	更快
返回标量	scalar()	一值
返回单行对象	first() / one()
判断是否返回行	returns_rows
UPDATE 影响行数	rowcount

如何让 Result 能多次调用

在 SQLAlchemy 2.0 中，Result（以及 ScalarResult）默认是一次性可迭代（consumable iterator），意味着：

你一旦调用 .all()、.first()、.fetchmany()、.scalars()、迭代 for row in result —— 结果就被“消费掉”；
之后就不能再次迭代或重新读取。

所以你看到的现象是：

Result 有时只能用一次，再次使用就为空或报错

✅ 如何让 Result 能多次调用？

方法 1：一次性读取到内存，再复用（最常见）

你可以立即把结果 .all() 进列表，然后随便用多少次：

result = session.execute(stmt)
rows = result.all()   # List[Row]，可无限次使用

# 多次使用
print(rows)
print(rows[0])

🌟🌟 如果你需要字典： 🌟🌟

dicts = [dict(row) for row in result.mappings().all()]

✅ 最简单、最安全，也最推荐的方式

方法 2：使用 session.execute(…).mappings() 转换成可重复的结构

result = session.execute(stmt).mappings().all()

得到的是：

[{‘id’: 1, ’name’: ‘foo’}, {‘id’: 2, ’name’: ‘bar’}]

一样可以多次使用。

方法 3：让查询返回 ORM 对象（ORM 模式）

使用：

rows = session.scalars(select(User)).all()

rows 是 list[User]，自然可反复使用。

方法 4：重新执行查询（如果你不能缓存数据）

res1 = session.execute(stmt).all()
res2 = session.execute(stmt).all()

这种方式比较笨但可行。

❌ 不可行：Result 本身不能重复使用

官方说明（简化）：

Result 是一次性游标，不保证多次迭代。

因此不存在一种方法让 Result 本身变成 “可反复读取” 的对象。

🔥 那到底应该怎么做？（最佳实践）

写一个通用的封装：永远一次性 consume 后返回 list

例如：

def query_all(session, stmt):
    result = session.execute(stmt)
    return result.all()

或者字典版本：

def query_dicts(session, stmt):
    return [dict(row) for row in session.execute(stmt).mappings().all()]

你的业务代码永远拿到的是：

list
可 JSON 序列化
可复用

🎯 最推荐的方式

如果你想 ORM 对象：

rows = session.scalars(stmt).all()

如果你想 dict：

rows = session.execute(stmt).mappings().all()

二者都返回可以反复使用的 Python 列表。

⭐ 总结

方式	Result 可多次使用？	推荐？
直接使用 Result	❌ 不能	不推荐
使用 .all() 转列表	✅ 能	★★★★★
使用 .mappings().all()	✅ 能	★★★★★
使用 scalars().all()	✅ 能	★★★★★
重新执行查询	✅ 能（但低效）	★★☆☆☆

Result 示例

test_result_01.py

from session import db_session
from sqlalchemy import text

if __name__ == "__main__":

    with db_session() as session:

        # 单列多行
        sql = text("SELECT id, trim(username) as username FROM users;")

        # ----------------------------------------------------------------------

        print(session.execute(sql).all())
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # ----------------------------------------------------------------------

        print(session.execute(sql).columns("username"))
        # <sqlalchemy.engine.cursor.CursorResult object at 0x106ee4f30>

        # 不推荐
        print(list(session.execute(sql).columns("username")))
        # [('admin',), ('septvean',), ('smalloc',)]

        # 推荐
        print(session.execute(sql).columns("username").all())
        # [('admin',), ('septvean',), ('smalloc',)]

        for username in session.execute(sql).columns("username"):
            print(username)

            # ('admin',)
            # ('septvean',)
            # ('smalloc',)

        # ----------------------------------------------------------------------

        print(session.execute(sql).fetchall())
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        print(session.execute(sql).fetchmany(2))
        # [(1, 'admin'), (2, 'septvean')]

        print(session.execute(sql).fetchone())
        # (1, 'admin')

        print(session.execute(sql).first())
        # (1, 'admin')

        # ----------------------------------------------------------------------

        print(session.execute(sql).keys())
        # RMKeyView(['id', 'username'])

        print(tuple(session.execute(sql).keys()))
        # ('id', 'username')

        # ----------------------------------------------------------------------

        print(session.execute(sql).mappings())
        # <sqlalchemy.engine.result.MappingResult object at 0x10611dd60>

        # 不推荐
        print(list(session.execute(sql).mappings()))
        # [{'id': 1, 'username': 'admin'}, {'id': 2, 'username': 'septvean'}, {'id': 4, 'username': 'smalloc'}]

        # 推荐
        print(session.execute(sql).mappings().all())
        # [{'id': 1, 'username': 'admin'}, {'id': 2, 'username': 'septvean'}, {'id': 4, 'username': 'smalloc'}]

        # ----------------------------------------------------------------------

        # print(session.execute(sql).one())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

        # print(session.execute(sql).one_or_none())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when one or none was required

        # ----------------------------------------------------------------------

        print(session.execute(sql).partitions(2))
        # <generator object Result.partitions at 0x105830900>

        print(tuple(session.execute(sql).partitions(2)))
        # ([(1, 'admin'), (2, 'septvean')], [(4, 'smalloc')])

        # ----------------------------------------------------------------------

        print(session.execute(sql).scalar())
        # 1

        # print(session.execute(sql).scalar_one())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

        # print(session.execute(sql).scalar_one_or_none())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when one or none was required

        # ----------------------------------------------------------------------

        print(session.execute(sql).scalars())
        # <sqlalchemy.engine.result.ScalarResult object at 0x1079569e0>

        # 不推荐
        print(list(session.execute(sql).scalars()))
        # [1, 2, 4]

        # 推荐
        print(session.execute(sql).scalars().all())
        # [1, 2, 4]

        # ----------------------------------------------------------------------

        print(session.execute(sql).tuples())
        # <sqlalchemy.engine.cursor.CursorResult object at 0x106dc50f0>

        # 不推荐
        print(list(session.execute(sql).tuples()))
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # 推荐
        print(session.execute(sql).tuples().all())
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # ----------------------------------------------------------------------

        print(session.execute(sql).unique())
        # <sqlalchemy.engine.cursor.CursorResult object at 0x103bb50f0>

        # 不推荐
        print(list(session.execute(sql).unique()))
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # 推荐
        print(session.execute(sql).unique().all())
        # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

test_result_02.py

from database import session_execute
from session import db_session
from sqlalchemy import text

if __name__ == "__main__":

    with db_session() as session:

        # 单列多行
        sql = text("SELECT id, trim(username) as username FROM users;")

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.all())
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.columns("username"))
            # <sqlalchemy.engine.cursor.CursorResult object at 0x10844c830>

        # 不推荐
        if result := session_execute(session, sql):
            print(list(result.columns("username")))
            # [('admin',), ('septvean',), ('smalloc',)]

        # 推荐
        if result := session_execute(session, sql):
            print(result.columns("username").all())
            # [('admin',), ('septvean',), ('smalloc',)]

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.fetchall())
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        if result := session_execute(session, sql):
            print(result.fetchmany(2))
            # [(1, 'admin'), (2, 'septvean')]

        if result := session_execute(session, sql):
            print(result.fetchone())
            # (1, 'admin')

        if result := session_execute(session, sql):
            print(result.first())
            # (1, 'admin')

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.keys())
            # RMKeyView(['id', 'username'])

        if result := session_execute(session, sql):
            print(tuple(result.keys()))
            # ('id', 'username')

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.mappings())
            # <sqlalchemy.engine.result.MappingResult object at 0x10847bac0>

        # 不推荐
        if result := session_execute(session, sql):
            print(list(result.mappings()))
            # [{'id': 1, 'username': 'admin'}, {'id': 2, 'username': 'septvean'}, {'id': 4, 'username': 'smalloc'}]

        # 推荐
        if result := session_execute(session, sql):
            print(result.mappings().all())
            # [{'id': 1, 'username': 'admin'}, {'id': 2, 'username': 'septvean'}, {'id': 4, 'username': 'smalloc'}]

        # ----------------------------------------------------------------------

        # if result := session_execute(session, sql):
        #     print(result.one())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

        # if result := session_execute(session, sql):
        #     print(result.one_or_none())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when one or none was required

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.partitions(2))
            # <generator object Result.partitions at 0x105d73d80>

        if result := session_execute(session, sql):
            print(tuple(result.partitions(2)))
            # ([(1, 'admin'), (2, 'septvean')], [(4, 'smalloc')])

        # ----------------------------------------------------------------------

        # if result := session_execute(session, sql):
        #     print(result.scalar())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

        # if result := session_execute(session, sql):
        #     print(result.scalar_one())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when exactly one was required

        # if result := session_execute(session, sql):
        #     print(result.scalar_one_or_none())
        # sqlalchemy.exc.MultipleResultsFound: Multiple rows were found when one or none was required

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.scalars())
            # <sqlalchemy.engine.result.ScalarResult object at 0x105b3fa70>

        # 不推荐
        if result := session_execute(session, sql):
            print(list(result.scalars()))
            # [1, 2, 4]

        # 推荐
        if result := session_execute(session, sql):
            print(result.scalars().all())
            # [1, 2, 4]

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.tuples())
            # <sqlalchemy.engine.cursor.CursorResult object at 0x105b108a0>

        # 不推荐
        if result := session_execute(session, sql):
            print(list(result.tuples()))
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # 推荐
        if result := session_execute(session, sql):
            print(result.tuples().all())
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # ----------------------------------------------------------------------

        if result := session_execute(session, sql):
            print(result.unique())
            # <sqlalchemy.engine.cursor.CursorResult object at 0x105b10980>

        # 不推荐
        if result := session_execute(session, sql):
            print(list(result.unique()))
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

        # 推荐
        if result := session_execute(session, sql):
            print(result.unique().all())
            # [(1, 'admin'), (2, 'septvean'), (4, 'smalloc')]

ScalarResult 示例

Result 可以通过 scalars()方法转换为 ScalarResult。

test_scalar_result_01.py

from database import session_execute, session_scalars
from session import db_session
from sqlalchemy import text

if __name__ == "__main__":

    with db_session() as session:

        # 单列单行 (单个)
        sql = text("SELECT COUNT(*) AS total FROM users;")

        # ----------------------------------------------------------------------

        # execute + scalar_one_or_none
        # 也可以写成一行
        r1 = session.execute(sql)
        print(r1.scalar_one_or_none())  # 3
        print(session.execute(sql).scalar_one_or_none())  # 3

        # ----------------------------------------------------------------------

        # 注意:
        #
        #   如果调用了 scalar_one_or_none() 方法, 就不能再调用 scalars()
        #   否则会报错: This result object is closed
        #
        #   print(r1.scalar_one_or_none())
        #   # 3
        #
        #   print(r1.scalars().one_or_none())
        #   # sqlalchemy.exc.ResourceClosedError: This result object is closed.
        #
        # scalars() 实际上是将 Result 转换为 ScalarResult

        # ----------------------------------------------------------------------

        # execute + scalars
        r2 = session.execute(sql)
        r2 = r2.scalars()
        print(r2.one_or_none())  # 3

        # 或者
        r2 = session.execute(sql).scalars()
        print(r2.one_or_none())  # 3

        # 也可以写成一行
        print(session.execute(sql).scalars().one_or_none())  # 3

        # ----------------------------------------------------------------------

        # scalars + one_or_none
        # 也可以写成一行
        r3 = session.scalars(sql)
        print(r3.one_or_none())  # 3
        print(session.scalars(sql).one_or_none())  # 3

        # ----------------------------------------------------------------------

        # session_execute + scalar_one_or_none
        r4 = session_execute(session, sql)
        if r4 is not None:
            print(r4.scalar_one_or_none())  # 3

        # ----------------------------------------------------------------------

        # session_scalars + one_or_none
        r5 = session_scalars(session, sql)
        if r5 is not None:
            print(r5.one_or_none())  # 3

test_scalar_result_02.py

from database import session_scalars
from session import db_session
from sqlalchemy import text

if __name__ == "__main__":

    with db_session() as session:

        # 单列多行
        sql = text("SELECT trim(username) FROM users;")

        # ----------------------------------------------------------------------

        # 常规

        print(session.scalars(sql).all())  # ['admin', 'septvean', 'smalloc']
        print(session.scalars(sql).fetchmany(2))  # ['admin', 'septvean']
        print(session.scalars(sql).first())  # admin

        # ----------------------------------------------------------------------

        # 去重

        print(session.scalars(sql).unique().all())  # ['admin', 'septvean', 'smalloc']
        print(session.scalars(sql).unique().fetchmany(2))  # ['admin', 'septvean']
        print(session.scalars(sql).unique().first())  # admin

        # ----------------------------------------------------------------------

        # 常规

        if result := session_scalars(session, sql):
            print(result.all())  # ['admin', 'septvean', 'smalloc']

        if result := session_scalars(session, sql):
            print(result.fetchmany(2))  # ['admin', 'septvean']

        if result := session_scalars(session, sql):
            print(result.first())  # admin

        # ----------------------------------------------------------------------

        # 去重

        if result := session_scalars(session, sql):
            print(result.unique().all())  # ['admin', 'septvean', 'smalloc']

        if result := session_scalars(session, sql):
            print(result.unique().fetchmany(2))  # ['admin', 'septvean']

        if result := session_scalars(session, sql):
            print(result.unique().first())  # admin

推导数据

results = ["admin", "septvean", "smalloc"]

users = [{"username": item} for item in results]

# [{"username": "admin"}, {"username": "septvean"}, {"username": "smalloc"}]