Tortoise ORM 查询操作完全指南

官方文档：https://tortoise.github.io/query.html
适用版本：tortoise-orm 0.20+（2026-05-08 核实）
最后更新：2026-03-05

QuerySet 基础概念

• QuerySet 是惰性的：构建时不发 SQL，await 时才执行
• 大多数方法返回新的 QuerySet（链式调用）
• 最终执行方法：await qs、.all()、.get()、.count() 等

# 链式构建
qs = User.filter(is_active=True).order_by('-created_at').limit(10)
users = await qs   # 此时才发 SQL

一、基础查询

Model.all() -> QuerySet

返回查询所有记录的 QuerySet（不立即执行）。

users = await User.all()

Model.filter(*args, **kwargs) -> QuerySet

过滤条件，支持 Django 风格的 lookup。所有条件默认 AND 连接。

# 简单相等过滤
users = await User.filter(is_active=True, age=18)

# lookup 表达式
users = await User.filter(name__contains='alice')
users = await User.filter(age__gte=18, age__lt=60)

完整 Lookup 列表：

日期时间专用 Lookup（PostgreSQL/MySQL）：

await Order.filter(created_at__year=2024, created_at__month=3)

Model.exclude(*args, **kwargs) -> QuerySet

排除匹配条件的记录，等价于 NOT (条件)。

active_users = await User.exclude(is_deleted=True)

Model.get(**kwargs) -> Model

获取恰好一条记录，找不到抛 DoesNotExist，多条抛 MultipleObjectsReturned。

try:
    user = await User.get(id=1)
    user = await User.get(email='[email protected]', is_active=True)
except User.DoesNotExist:
    ...

Model.get_or_none(**kwargs) -> Model | None

找不到返回 None，找到多条仍抛 MultipleObjectsReturned。

user = await User.get_or_none(id=999)  # None，不报错

QuerySet.first() -> Model | None

返回第一条记录（按当前排序），无结果返回 None。

latest = await User.filter(is_active=True).order_by('-created_at').first()

QuerySet.last() -> Model | None (需要先 order_by)

返回最后一条记录。

QuerySet.earliest(*orderings) -> Model | None

按指定字段升序取第一条（等价于 .order_by(*orderings).first()）。

oldest = await User.all().earliest('created_at')

QuerySet.latest(*orderings) -> Model | None

按指定字段降序取第一条（等价于 .order_by(*['-' + f for f in orderings]).first()）。

newest = await User.all().latest('created_at')

二、创建/保存

Model.create(**kwargs) -> Model

创建并立即保存到数据库。

user = await User.create(
    username='alice',
    email='[email protected]',
    is_active=True,
)

Model.get_or_create(defaults=None, **kwargs) -> tuple[Model, bool]

返回 (instance, created)，created 为 True 表示是新创建的。

user, created = await User.get_or_create(
    defaults={'is_active': True, 'role': 'user'},
    email='[email protected]',
)

Model.update_or_create(defaults=None, **kwargs) -> tuple[Model, bool]

存在则更新（用 defaults），不存在则创建。参数同 get_or_create。

user, created = await User.update_or_create(
    defaults={'name': 'Alice Updated'},
    email='[email protected]',
)

instance.save(update_fields=None, force_create=False, force_update=False, using_db=None)

user.name = 'Bob'
await user.save()

# 只更新 name 字段
await user.save(update_fields=['name'])

Model.bulk_create(objects, batch_size=None, ignore_conflicts=False, update_fields=None, on_conflict=None) -> list[Model]

users = [User(username=f'user_{i}') for i in range(1000)]
await User.bulk_create(users, batch_size=100)

# Upsert（MySQL 8+ / PostgreSQL / SQLite 3.24+）
await User.bulk_create(
    users,
    on_conflict=['email'],
    update_fields=['name', 'updated_at'],
)

Model.bulk_update(objects, fields, batch_size=None)

users = await User.filter(is_active=False).all()
for u in users:
    u.is_active = True
await User.bulk_update(users, fields=['is_active'], batch_size=100)

三、更新/删除

QuerySet.update(**kwargs) -> int

批量更新，返回受影响行数。

count = await User.filter(is_active=False).update(is_deleted=True)
print(f'更新了 {count} 条')

注意：update() 不会触发 auto_now 字段更新（直接发 SQL UPDATE，不经过 Python 层）。需要更新时间戳时要显式传入：

from datetime import datetime, timezone
await User.filter(...).update(
    status='inactive',
    updated_at=datetime.now(timezone.utc)
)

QuerySet.delete() -> tuple[int, dict]

批量删除，返回 (删除总数, 按类型分布)。

deleted_count, _ = await User.filter(is_deleted=True).delete()

instance.delete()

删除单条记录。

user = await User.get(id=1)
await user.delete()

四、排序/分页

QuerySet.order_by(*fields) -> QuerySet

users = await User.all().order_by('-created_at', 'username')

QuerySet.limit(n) -> QuerySet

限制返回条数。

QuerySet.offset(n) -> QuerySet

跳过前 n 条。

# 分页（第2页，每页10条）
page, size = 2, 10
users = await User.all().order_by('id').offset((page - 1) * size).limit(size)

五、聚合/统计

QuerySet.count() -> int

统计数量（发 SELECT COUNT(*)）。

total = await User.filter(is_active=True).count()

QuerySet.exists() -> bool

判断是否存在匹配记录（比 count() > 0 更高效，用 EXISTS 子查询）。

if await User.filter(email='[email protected]').exists():
    raise ValueError('邮箱已存在')

QuerySet.annotate(**kwargs) -> QuerySet

添加聚合注解字段，需配合 functions 使用。

from tortoise.functions import Count, Sum, Avg, Max, Min

# 统计每个部门的员工数
depts = await Department.annotate(
    emp_count=Count('employees')
).all()
for dept in depts:
    print(dept.name, dept.emp_count)

# 聚合 + 过滤（HAVING）
depts = await Department.annotate(
    emp_count=Count('employees')
).filter(emp_count__gte=10)

常用聚合函数：

直接聚合（不用 annotate）

from tortoise.expressions import F
from tortoise.functions import Sum

# 单一聚合值（需用 values/values_list + annotate）
result = await Order.annotate(total=Sum('amount')).values('total')
total_amount = result[0]['total']

六、字段选择

QuerySet.values(*fields) -> QuerySet[dict]

只返回指定字段，结果为字典列表。

users = await User.all().values('id', 'username', 'email')
# [{'id': 1, 'username': 'alice', 'email': '[email protected]'}, ...]

# 关联字段（用 __ 跨越关联）
students = await Student.all().values('id', 'name', 'clazz__name')

QuerySet.values_list(*fields, flat=False) -> QuerySet[tuple|scalar]

ids = await User.all().values_list('id', flat=True)
# [1, 2, 3, ...]

pairs = await User.all().values_list('id', 'username')
# [(1, 'alice'), (2, 'bob'), ...]

QuerySet.distinct() -> QuerySet

对结果去重（对应 SQL SELECT DISTINCT）。

cities = await User.all().distinct().values_list('city', flat=True)

QuerySet.group_by(*fields) -> QuerySet

分组查询（对应 SQL GROUP BY），通常配合 annotate 使用。

from tortoise.functions import Count

result = await Order.group_by('status').annotate(count=Count('id')).values('status', 'count')

QuerySet.only(*fields) -> QuerySet

只 SELECT 指定字段，返回的模型实例中其余字段不加载（访问会发额外查询）。比 values() 更灵活——返回的仍是模型实例。

users = await User.all().only('id', 'username')

QuerySet.in_bulk(id_list, field_name='pk') -> dict

根据 ID 列表批量查询，返回 {id: instance} 字典，便于快速查找。

user_map = await User.in_bulk([1, 2, 3])
# {1: <User 1>, 2: <User 2>, 3: <User 3>}
user = user_map.get(1)

七、复杂查询：Q 对象

用于构建 OR、NOT 等复杂逻辑。

from tortoise.expressions import Q

# OR 条件
users = await User.filter(
    Q(name='alice') | Q(name='bob')
)

# NOT 条件
users = await User.filter(~Q(is_deleted=True))

# AND + OR 混合
users = await User.filter(
    Q(is_active=True) & (Q(role='admin') | Q(role='moderator'))
)

# 嵌套 Q
q = Q(status='pending')
if start_date:
    q &= Q(created_at__gte=start_date)
users = await User.filter(q)

八、F 表达式

引用数据库字段本身，支持字段间运算（避免 Python 层的竞态）。

from tortoise.expressions import F

# 字段自增（原子操作）
await Product.filter(id=1).update(stock=F('stock') - 1)

# 字段间比较
await Order.filter(actual_amount__lt=F('expected_amount'))

# 注解中使用
from tortoise.functions import Coalesce
await Product.annotate(effective_price=Coalesce(F('sale_price'), F('price')))

九、预加载关联

详见 select_related 和 prefetch_related的区别和用法，此处只列参数。

QuerySet.select_related(*fields) -> QuerySet

JOIN 预加载 FK / O2O（正向）。

students = await Student.all().select_related('clazz', 'clazz__school')

QuerySet.prefetch_related(*fields) -> QuerySet

多 SQL 批量预加载反向关联 / M2M。

from tortoise.query_utils import Prefetch

classes = await Class.all().prefetch_related(
    Prefetch('students', queryset=Student.filter(is_active=True))
)

Prefetch 对象参数：

# 带过滤和排序的预加载
await Class.all().prefetch_related(
    Prefetch(
        'students',
        queryset=Student.filter(is_active=True).order_by('name'),
        to_attr='active_students'
    )
)
# 访问：class_obj.active_students

十、调试与原生 SQL

QuerySet.sql(params_inline=False) -> str

返回当前 QuerySet 生成的 SQL 字符串，不执行查询，用于调试。

qs = User.filter(is_active=True).order_by('-created_at').limit(10)
print(qs.sql())
# SELECT "id","username" FROM "user" WHERE "is_active"=? ORDER BY "created_at" DESC LIMIT 10

QuerySet.explain() -> str

执行 EXPLAIN 分析查询计划，返回数据库的查询执行计划文本。

plan = await User.filter(is_active=True).explain()
print(plan)

Tortoise.execute_query(query, values=None)

执行原生 SQL，返回 (rowcount, rows)。

from tortoise import Tortoise

conn = Tortoise.get_connection('default')
rows = await conn.execute_query('SELECT * FROM user WHERE id = $1', [user_id])

QuerySet.raw(sql) -> RawQuerySet

users = await User.raw('SELECT * FROM "user" WHERE is_active = 1')

建议优先使用 ORM 查询，原生 SQL 用于复杂统计报表场景。

十一、使用指定数据库连接

QuerySet.using_db(db) -> QuerySet

在多数据库配置时指定使用哪个连接。

users = await User.all().using_db('secondary')

完整使用示例

from tortoise.expressions import Q, F
from tortoise.functions import Count
from tortoise.query_utils import Prefetch

async def get_active_users_with_stats(page: int = 1, size: int = 20):
    offset = (page - 1) * size

    # 带统计、过滤、排序、分页的复杂查询
    users = await (
        User
        .filter(
            Q(is_active=True) & Q(created_at__year=2024)
        )
        .annotate(order_count=Count('orders'))
        .prefetch_related(
            Prefetch('orders', queryset=Order.filter(status='completed').order_by('-created_at'))
        )
        .order_by('-order_count', 'username')
        .offset(offset)
        .limit(size)
    )

    total = await User.filter(is_active=True, created_at__year=2024).count()

    return {'total': total, 'items': users}

常见踩坑

1. QuerySet 是惰性的：User.filter(...) 不执行查询，需要 await。
2. get() 的陷阱：条件不精确可能抛 MultipleObjectsReturned，不确定时用 get_or_none() 或 filter().first()。
3. update() 不触发 auto_now：批量 update 绕过了 Python 层，auto_now 字段不自动更新。
4. N+1 查询：循环中访问关联对象要用 select_related / prefetch_related 预加载。
5. count() vs exists()：只需判断是否存在用 exists()，更高效。
6. values() 返回字典：无法再调用 select_related()，需要实例时不用 values()。
7. 关联字段访问需 await：await student.clazz（如果未预加载）会单独发一条查询。

最佳实践

用 select_related / prefetch_related 预加载关联对象：在结果集上访问外键前声明预加载，消除 N+1：

students = await Student.all().select_related("clazz").prefetch_related("courses")
for s in students:
    print(s.clazz.name)  # 不额外发查询

exists() 优于 count() 做存在性判断：await User.filter(email=email).exists() 只需 SELECT 1，比 count() 快，避免全表计数。

update() 批量更新配合 using_db 参数：多数据库场景下显式指定连接，避免路由错误：

await User.filter(is_active=False).using_db("default").delete()

Q 对象构建复杂条件：OR 条件用 Q(field1=v1) | Q(field2=v2)，与简单 filter 的 AND 语义互补：

from tortoise.expressions import Q
await User.filter(Q(role="admin") | Q(is_superuser=True)).all()

分页用 offset() + limit()，不用 Python 切片：await queryset.offset(20).limit(10) 在 SQL 层分页；Python 切片 queryset[20:30] 在内存中切片（先全量加载）。

常见陷阱

陷阱：忘记 await QuerySet 不触发查询

现象： qs = User.filter(...) 后 qs 是 QuerySet 对象，而非结果列表，后续 len(qs) 或 qs[0] 报错。
原因： Tortoise ORM QuerySet 是惰性的，需要 await qs 或 await qs.values() 才执行 SQL。
解决： 始终 await 查询，或配合 async for item in qs 流式处理。

陷阱：get() 在多结果时抛 MultipleObjectsReturned

现象： 期望获取唯一记录，但数据中有多条匹配，运行时报 MultipleObjectsReturned。
原因： get() 要求结果恰好为一条，否则抛异常；零结果抛 DoesNotExist。
解决： 不确定唯一性时用 filter().first() 或 get_or_none()；仅在真正唯一约束字段（如 id、email unique）上用 get()。

陷阱：values() 结果无法访问关联字段

现象： queryset.values("name", "clazz__name") 报错或返回 None。
原因： values() 返回 dict，跨关联字段的访问语法取决于 Tortoise 版本，且无法再链式调用 select_related。
解决： 需要关联字段时改用 select_related + 普通实例列表；确实需要 dict 时用 values_list 或在 Pydantic schema 转换时处理。

参见

模型字段完全参考
事务与并发
初始化与配置