> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# Tortoise ORM 查询操作完全指南
- URL: https://blog.vercanti.com/tortoise-orm-cha-xun-cao-zuo-wan-quan-zhi-nan/
- Published: 2026-08-28T14:34:53.000Z
- Updated: 2026-08-28T14:57:35.000Z
- Description: 返回查询所有记录的 QuerySet（不立即执行）。 过滤条件，支持 Django 风格的 lookup。所有条件默认 AND 连接。 完整 Lookup 列表： 日期时间专用 Lookup（PostgreSQL/MySQL）： 排除匹配条件的记录，等价于 NOT (条件)。 获取恰好一条记录，找不到抛 DoesNotExist，多条抛 MultipleObjectsReturned。 找不到返回 None，找到多条仍抛 MultipleObjectsReturned。 返回第一条记录（按当前排序），无结果返回 None。 返回最后一条记录。 按指定字段
- Author: yellowdog
- Tags: 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()` 等

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

```

---

## 一、基础查询

### `Model.all() -> QuerySet`

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

```python
users = await User.all()

```

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

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

```python
# 简单相等过滤
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                 | 说明                                                | 示例                            |
| ---------------------- | ------------------------------------------------- | ----------------------------- |
| field / field\_\_exact | 精确匹配（默认）                                          | name='alice'                  |
| field\_\_iexact        | 大小写不敏感精确匹配                                        | name\_\_iexact='Alice'        |
| field\_\_contains      | 包含子串（LIKE %x%）                                    | name\_\_contains='ali'        |
| field\_\_icontains     | 大小写不敏感包含                                          | name\_\_icontains='Ali'       |
| field\_\_startswith    | 前缀匹配（LIKE x%）                                     | name\_\_startswith='A'        |
| field\_\_istartswith   | 大小写不敏感前缀                                          |                               |
| field\_\_endswith      | 后缀匹配（LIKE %x）                                     | name\_\_endswith='.py'        |
| field\_\_iendswith     | 大小写不敏感后缀                                          |                               |
| field\_\_in            | 在列表中                                              | id\_\_in=\[1,2,3\]            |
| field\_\_not\_in       | 不在列表中                                             | id\_\_not\_in=\[1,2,3\]       |
| field\_\_gt            | 大于                                                | age\_\_gt=18                  |
| field\_\_gte           | 大于等于                                              | age\_\_gte=18                 |
| field\_\_lt            | 小于                                                | age\_\_lt=60                  |
| field\_\_lte           | 小于等于                                              |                               |
| field\_\_range         | 范围（含两端）                                           | age\_\_range=(18, 60)         |
| field\_\_isnull        | 是否为 NULL                                          | deleted\_at\_\_isnull=True    |
| field\_\_not           | 不等于                                               | status\_\_not='deleted'       |
| field\_\_not\_isnull   | 非 NULL                                            | email\_\_not\_isnull=True     |
| field\_\_search        | 全文搜索（PostgreSQL 使用 plainto\_tsquery，MySQL 使用全文索引） | content\_\_search='python'    |
| field\_\_posix\_regex  | POSIX 正则匹配（PostgreSQL/MySQL/SQLite）               | name\_\_posix\_regex='^A.\*'  |
| field\_\_iposix\_regex | 大小写不敏感 POSIX 正则（PostgreSQL/SQLite，不支持 MySQL）      | name\_\_iposix\_regex='^a.\*' |

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

| Lookup               | 说明          |
| -------------------- | ----------- |
| field\_\_year        | 年份等于        |
| field\_\_quarter     | 季度等于（1-4）   |
| field\_\_month       | 月份等于        |
| field\_\_week        | 周数等于（ISO 周） |
| field\_\_day         | 日等于         |
| field\_\_hour        | 小时等于        |
| field\_\_minute      | 分钟等于        |
| field\_\_second      | 秒等于         |
| field\_\_microsecond | 微秒等于        |

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

```

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

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

```python
active_users = await User.exclude(is_deleted=True)

```

### `Model.get(**kwargs) -> Model`

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

```python
try:
    user = await User.get(id=1)
    user = await User.get(email='a@b.com', is_active=True)
except User.DoesNotExist:
    ...

```

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

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

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

```

### `QuerySet.first() -> Model | None`

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

```python
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()`）。

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

```

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

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

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

```

---

## 二、创建/保存

### `Model.create(**kwargs) -> Model`

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

```python
user = await User.create(
    username='alice',
    email='alice@example.com',
    is_active=True,
)

```

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

| 参数         | 说明                   |
| ---------- | -------------------- |
| defaults   | 仅在创建时使用的字段值（不参与查找条件） |
| \*\*kwargs | 查找条件                 |

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

```python
user, created = await User.get_or_create(
    defaults={'is_active': True, 'role': 'user'},
    email='alice@example.com',
)

```

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

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

```python
user, created = await User.update_or_create(
    defaults={'name': 'Alice Updated'},
    email='alice@example.com',
)

```

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

| 参数             | 说明                    |
| -------------- | --------------------- |
| update\_fields | 列表，只更新指定字段（节省 SQL 带宽） |
| force\_create  | 强制 INSERT（即使有 pk）     |
| force\_update  | 强制 UPDATE             |
| using\_db      | 指定数据库连接               |

```python
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]`

| 参数                | 说明                                               |
| ----------------- | ------------------------------------------------ |
| objects           | Model 实例列表                                       |
| batch\_size       | 每批次插入数量，None 表示全部一批                              |
| ignore\_conflicts | True 时冲突行被跳过（不抛出异常）                              |
| update\_fields    | 冲突时更新的字段（ON CONFLICT DO UPDATE，需配合 on\_conflict） |
| on\_conflict      | 冲突检测字段（用于 upsert）                                |

```python
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)`

| 参数          | 说明              |
| ----------- | --------------- |
| objects     | 已修改的 Model 实例列表 |
| fields      | 要更新的字段名列表       |
| batch\_size | 每批次数量           |

```python
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`

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

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

```

> 注意：`update()` 不会触发 `auto_now` 字段更新（直接发 SQL UPDATE，不经过 Python 层）。需要更新时间戳时要显式传入：
> 
> ```python
> from datetime import datetime, timezone
> await User.filter(...).update(
>    status='inactive',
>    updated_at=datetime.now(timezone.utc)
> )
> 
> ```

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

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

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

```

### `instance.delete()`

删除单条记录。

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

```

---

## 四、排序/分页

### `QuerySet.order_by(*fields) -> QuerySet`

| 参数格式               | 说明                           |
| ------------------ | ---------------------------- |
| 'field'            | 升序                           |
| '-field'           | 降序（前缀 \-）                    |
| 'related\_\_field' | 关联模型字段排序（需先 select\_related） |

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

```

### `QuerySet.limit(n) -> QuerySet`

限制返回条数。

### `QuerySet.offset(n) -> QuerySet`

跳过前 n 条。

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

```

---

## 五、聚合/统计

### `QuerySet.count() -> int`

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

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

```

### `QuerySet.exists() -> bool`

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

```python
if await User.filter(email='a@b.com').exists():
    raise ValueError('邮箱已存在')

```

### `QuerySet.annotate(**kwargs) -> QuerySet`

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

```python
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)

```

**常用聚合函数：**

| 函数                           | 说明                  |
| ---------------------------- | ------------------- |
| Count(field, distinct=False) | 计数；distinct=True 去重 |
| Sum(field)                   | 求和                  |
| Avg(field)                   | 平均值                 |
| Max(field)                   | 最大值                 |
| Min(field)                   | 最小值                 |
| Concat(field1, field2, ...)  | 字符串拼接               |
| Coalesce(field, default)     | 第一个非 NULL 值         |
| Length(field)                | 字符串长度               |
| Trim(field)                  | 去除首尾空格              |

### 直接聚合（不用 annotate）

```python
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]`

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

```python
users = await User.all().values('id', 'username', 'email')
# [{'id': 1, 'username': 'alice', 'email': 'a@b.com'}, ...]

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

```

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

| 参数       | 说明                         |
| -------- | -------------------------- |
| \*fields | 要获取的字段名                    |
| flat     | True 且只传一个字段时，返回标量列表而非元组列表 |

```python
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`）。

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

```

### `QuerySet.group_by(*fields) -> QuerySet`

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

```python
from tortoise.functions import Count

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

```

### `QuerySet.only(*fields) -> QuerySet`

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

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

```

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

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

| 参数          | 说明            |
| ----------- | ------------- |
| id\_list    | ID 值列表        |
| field\_name | 用于查找的字段名，默认主键 |

```python
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 等复杂逻辑。

```python
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 层的竞态）。

```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的区别和用法](https://blog.vercanti.com/select%5Frelated-yu-prefetch%5Frelated-wan-quan-zhi-nan/)，此处只列参数。

### `QuerySet.select_related(*fields) -> QuerySet`

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

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

```

### `QuerySet.prefetch_related(*fields) -> QuerySet`

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

```python
from tortoise.query_utils import Prefetch

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

```

**`Prefetch` 对象参数：**

| 参数       | 说明                              |
| -------- | ------------------------------- |
| relation | 关联字段名                           |
| queryset | 自定义预加载查询（可带 filter、order\_by 等） |
| to\_attr | 挂载到对象的属性名（不指定则用 relation 名）     |

```python
# 带过滤和排序的预加载
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 字符串，不执行查询，用于调试。

| 参数             | 说明                                |
| -------------- | --------------------------------- |
| params\_inline | True 时将参数内联到 SQL 中，默认 False 使用占位符 |

```python
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` 分析查询计划，返回数据库的查询执行计划文本。

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

```

### `Tortoise.execute_query(query, values=None)`

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

```python
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`

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

```

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

---

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

### `QuerySet.using_db(db) -> QuerySet`

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

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

```

---

## 完整使用示例

```python
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：

```python
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` 参数**：多数据库场景下显式指定连接，避免路由错误：

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

```

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

```python
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 转换时处理。

---

## 参见

[模型字段完全参考](https://blog.vercanti.com/tortoise-orm-mo-xing-zi-duan-wan-quan-can-kao/)  
[事务与并发](https://blog.vercanti.com/tortoise-orm-shi-wu-yu-bing-fa/)  
[初始化与配置](https://blog.vercanti.com/tortoise-orm-chu-shi-hua-yu-pei-zhi/)