frappe-core-database

Frappe 数据库操作

快速参考

操作	方法	权限
获取文档	frappe.get_doc(doctype, name)	是
获取缓存文档	frappe.get_cached_doc(doctype, name)	否
新建文档	frappe.new_doc(doctype)	—
插入	doc.insert()	是
保存	doc.save()	是
删除文档	frappe.delete_doc(doctype, name)	是
列表（应用权限）	frappe.db.get_list(doctype, ...)	是
列表（不应用权限）	frappe.get_all(doctype, ...)	否
单个字段	frappe.db.get_value(doctype, name, field)	否
单个 DocType	frappe.db.get_single_value(doctype, field)	否
缓存值	frappe.db.get_value(..., cache=True)	否
直接更新	frappe.db.set_value(doctype, name, field, val)	否
直接更新	doc.db_set(field, value)	否
检查是否存在	frappe.db.exists(doctype, name)	否
计数	frappe.db.count(doctype, filters)	否
删除行	frappe.db.delete(doctype, filters)	否
原始 SQL	frappe.db.sql(query, values, as_dict)	否
查询构建器	frappe.qb.from_(doctype).select(...)	否

“权限” = 是 表示会自动应用用户权限筛选条件。

---

决策树

What do you need?
│
├─ Create / Update / Delete a document?
│  ├─ With validations + hooks → frappe.get_doc() + .insert()/.save()/.delete()
│  └─ Direct DB (no hooks) → frappe.db.set_value() or doc.db_set()
│
├─ Read a single document?
│  ├─ Need full object with methods → frappe.get_doc()
│  ├─ Read-only, rarely changes → frappe.get_cached_doc()
│  └─ Only need 1-2 fields → frappe.db.get_value()
│
├─ List of documents?
│  ├─ Respect user permissions → frappe.db.get_list()
│  └─ System/admin context → frappe.get_all()
│
├─ Single DocType value?
│  └─ frappe.db.get_single_value('Settings', 'field')
│
├─ Check existence?
│  └─ frappe.db.exists() — NEVER use get_doc in try/except
│
├─ Complex query (JOINs, aggregates)?
│  ├─ Cross-DB compatible → frappe.qb (Query Builder)
│  └─ DB-specific SQL → frappe.db.sql() with parameters
│
└─ DB-specific logic?
   └─ frappe.db.multisql({'mariadb': q1, 'postgres': q2})

规则：始终使用最高级别的抽象：ORM > Database API > Query Builder > Raw SQL。

---

ORM：文档操作

获取文档

doc = frappe.get_doc('Sales Invoice', 'SINV-00001')

# Single DocType (no name needed)
settings = frappe.get_doc('System Settings')

# Cached (read-only, for rarely-changing docs)
company = frappe.get_cached_doc('Company', 'My Company')

# Last created
last_task = frappe.get_last_doc('Task', filters={'status': 'Open'})

创建文档

doc = frappe.get_doc({
    'doctype': 'Task',
    'subject': 'Review report',
    'status': 'Open'
})
doc.insert()

# Alternative
doc = frappe.new_doc('Task')
doc.subject = 'Review report'
doc.insert()

更新文档

# Via ORM — triggers validate, on_update, etc.
doc = frappe.get_doc('Task', 'TASK-001')
doc.status = 'Completed'
doc.save()

# Direct DB — SKIPS all validations and hooks
frappe.db.set_value('Task', 'TASK-001', 'status', 'Completed')

# Direct DB on loaded doc
doc.db_set('status', 'Completed')
doc.db_set('status', 'Completed', update_modified=False)
doc.db_set({'status': 'Completed', 'priority': 'High'})

删除文档

frappe.delete_doc('Task', 'TASK-001')
# Also removes linked Communications, Comments, etc.

插入标志

doc.insert(
    ignore_permissions=True,     # Bypass permission check
    ignore_links=True,           # Skip link validation
    ignore_if_duplicate=True,    # No error on duplicate
    ignore_mandatory=True        # Skip required field check
)

规则：除非有书面记录的理由，否则绝不要同时使用多个忽略标志。每添加一个标志，都会削弱数据完整性。

---

Database API：读取

get_value

# Single field → scalar
status = frappe.db.get_value('Task', 'TASK-001', 'status')

# Multiple fields → tuple
subject, status = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'])

# As dict
data = frappe.db.get_value('Task', 'TASK-001', ['subject', 'status'], as_dict=True)

# With filters instead of name
status = frappe.db.get_value('Task', {'project': 'PROJ-001'}, 'status')

# Cached (for values that rarely change)
country = frappe.db.get_value('Company', 'MyCompany', 'country', cache=True)

get_single_value

timezone = frappe.db.get_single_value('System Settings', 'time_zone')

get_list / get_all

# get_list — applies user permissions
tasks = frappe.db.get_list('Task',
    filters={'status': 'Open'},
    fields=['name', 'subject', 'assigned_to'],
    order_by='creation desc',
    start=0,
    page_length=50
)

# get_all — NO permission check (same API, different default)
all_tasks = frappe.get_all('Task', filters={'status': 'Open'})

# pluck — returns flat list of single field
names = frappe.get_all('Task', filters={'status': 'Open'}, pluck='name')
# Returns: ['TASK-001', 'TASK-002', ...]

exists / count

exists = frappe.db.exists('User', '[email protected]')
exists = frappe.db.exists('User', {'email': '[email protected]'})

total = frappe.db.count('Task')
open_count = frappe.db.count('Task', {'status': 'Open'})

---

筛选运算符

{'status': 'Open'}                                    # =
{'status': ['!=', 'Cancelled']}                       # !=
{'amount': ['>', 1000]}                               # >
{'amount': ['>=', 1000]}                              # >=
{'status': ['in', ['Open', 'Working']]}               # IN
{'status': ['not in', ['Cancelled', 'Closed']]}       # NOT IN
{'date': ['between', ['2024-01-01', '2024-12-31']]}   # BETWEEN
{'subject': ['like', '%urgent%']}                      # LIKE
{'description': ['is', 'set']}                         # IS NOT NULL
{'description': ['is', 'not set']}                     # IS NULL

组合筛选条件

# AND — all conditions in one dict
filters = {'status': 'Open', 'priority': 'High'}

# AND — list format (allows duplicate fields)
filters = [['status', '=', 'Open'], ['priority', '=', 'High']]

# OR — separate parameter
or_filters = {'priority': 'Urgent', 'status': 'Overdue'}

---

Database API：写入

set_value

# Single field
frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed')

# Multiple fields
frappe.db.set_value('Task', 'TASK-001', {'status': 'Closed', 'priority': 'Low'})

# Without updating modified timestamp
frappe.db.set_value('Task', 'TASK-001', 'status', 'Closed', update_modified=False)

delete / truncate

# Delete with filters (DML — can be rolled back)
frappe.db.delete('Error Log', {'creation': ['<', '2024-01-01']})

# Truncate (DDL — CANNOT be rolled back)
frappe.db.truncate('Error Log')

bulk_update [v15+]

frappe.db.bulk_update('Task', {
    'TASK-001': {'status': 'Closed'},
    'TASK-002': {'status': 'Closed'}
}, chunk_size=100)

---

原始 SQL：始终使用参数化查询

# ✅ CORRECT — parameterized query
results = frappe.db.sql("""
    SELECT name, subject FROM `tabTask`
    WHERE status = %(status)s AND owner = %(owner)s
""", {'status': 'Open', 'owner': frappe.session.user}, as_dict=True)

严重：绝不要在 SQL 中使用 f 字符串、% 格式化或字符串拼接。请参阅 SQL 注入防护。

返回类型

frappe.db.sql(query)                    # Tuple of tuples (default)
frappe.db.sql(query, as_dict=True)      # List of dicts
frappe.db.sql(query, as_list=True)      # List of lists

表命名

始终使用反引号括起来的 tab 前缀：`tabSales Invoice`、`tabTask`

数据库专用 SQL

frappe.db.multisql({
    'mariadb': "SELECT IFNULL(field, 0) FROM `tabDoc`",
    'postgres': "SELECT COALESCE(field, 0) FROM `tabDoc`"
})

---

查询构建器（frappe.qb）[v14+]

查询构建器底层使用 PyPika。它会自动生成参数化 SQL。

Task = frappe.qb.DocType('Task')

results = (
    frappe.qb.from_(Task)
    .select(Task.name, Task.subject, Task.status)
    .where(Task.status == 'Open')
    .orderby(Task.creation, order='desc')
    .limit(10)
).run(as_dict=True)

JOIN

SI = frappe.qb.DocType('Sales Invoice')
Customer = frappe.qb.DocType('Customer')

results = (
    frappe.qb.from_(SI)
    .inner_join(Customer).on(SI.customer == Customer.name)
    .select(SI.name, SI.grand_total, Customer.customer_name)
    .where(SI.docstatus == 1)
).run(as_dict=True)

聚合函数

from frappe.query_builder.functions import Count, Sum, Avg

stats = (
    frappe.qb.from_(Task)
    .select(Task.status, Count(Task.name).as_('count'))
    .groupby(Task.status)
).run(as_dict=True)

OR 条件

customers = frappe.qb.DocType('Customer')
results = (
    frappe.qb.from_(customers)
    .select(customers.name)
    .where(
        (customers.territory == 'US') | (customers.territory == 'UK')
    )
).run(as_dict=True)

检查生成的 SQL

query = frappe.qb.from_(Task).select('*').where(Task.name == 'X')
sql, params = query.walk()   # Returns (sql_string, param_dict)
sql_str = query.get_sql()    # Returns SQL string

子查询、ImportMapper、ConstantColumn 和自定义函数请参阅 references/query-patterns.md。

---

缓存

文档缓存

doc = frappe.get_cached_doc('Company', 'My Company')   # Full document
val = frappe.db.get_value('Company', 'X', 'country', cache=True)  # Single value

Redis 缓存

frappe.cache.set_value('key', data, expires_in_sec=3600)
data = frappe.cache.get_value('key')
frappe.cache.delete_value('key')

@redis_cache 装饰器

from frappe.utils.caching import redis_cache

@redis_cache(ttl=300)
def get_dashboard_data(user):
    return expensive_calculation(user)

# Invalidate
get_dashboard_data.clear_cache()

哈希操作、失效策略和最佳实践请参阅 references/caching-patterns.md。

---

事务管理

框架会自动管理事务：

上下文	提交	回滚
POST/PUT 请求	成功后	未捕获异常时
GET 请求	从不	—
后台任务	成功后	发生异常时
补丁	成功后	发生异常时

手动事务（很少需要）

frappe.db.savepoint('before_payment')
try:
    # operations...
    frappe.db.commit()
except Exception:
    frappe.db.rollback(save_point='before_payment')

事务钩子 [v15+]

frappe.db.after_commit.add(sync_to_external_system)
frappe.db.after_rollback.add(cleanup_external_state)

---

SQL 注入防护

严重安全规则：绝不要将用户输入插入 SQL 字符串。

# ❌ VULNERABLE — SQL injection risk
frappe.db.sql(f"SELECT * FROM `tabUser` WHERE name = '{user_input}'")
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = '%s'" % user_input)
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = " + user_input)

# ✅ SAFE — parameterized query
frappe.db.sql("SELECT * FROM `tabUser` WHERE name = %(name)s", {'name': user_input})

# ✅ SAFE — ORM / Query Builder (always parameterized)
frappe.get_all('User', filters={'name': user_input})

User = frappe.qb.DocType('User')
frappe.qb.from_(User).select('*').where(User.name == user_input).run()

规则：必须使用 frappe.db.sql() 时，始终使用 %(param)s 占位符和字典。对于新代码，始终优先使用查询构建器（frappe.qb），而非原始 SQL。

---

版本差异

功能	v14	v15	v16
查询构建器（frappe.qb）	是	是	是
事务钩子	否	是	是
bulk_update	否	是	是
run=False 返回值	SQL 字符串	SQL 字符串	查询构建器对象
聚合字段语法	字符串	字符串	字典

v16 破坏性变更

# v14/v15 — string aggregates
fields=['count(name) as count']

# v16 — dict aggregates
fields=[{'COUNT': 'name', 'as': 'count'}]

# v14/v15 — run=False returns SQL string
sql = frappe.db.get_list('Task', run=False)

# v16 — run=False returns Query Builder object
qb_obj = frappe.db.get_list('Task', run=False)
sql = qb_obj.get_sql()

---

关键规则总结

1. 绝不在 SQL 中使用字符串格式化——始终使用参数化查询
2. 绝不在控制器钩子（validate、on_update 等）中调用 frappe.db.commit()
3. 始终为列表查询分页——使用 page_length 参数
4. 始终指定字段——在生产环境中绝不使用 fields=['*']
5. 始终使用 frappe.db.exists() 检查是否存在——绝不尝试通过 get_doc 配合 try/except 实现
6. 始终在原始 SQL 中为表名添加 tab 前缀：`tabSales Invoice`
7. 绝不在没有文档说明的理由时使用多个忽略标志
8. 始终使用批量获取，避免 N+1 查询
9. 对于新代码，始终优先使用 frappe.qb，而不是 frappe.db.sql()
10. 绝不在不了解 frappe.db.truncate() 无法回滚的情况下使用它

---

查询构建器：专用技能

对于复杂查询（联接、聚合、子查询、跨数据库兼容性），请参阅 frappe-syntax-query-builder。

• frappe.db 方法（此技能）——简单的 CRUD、get_value、get_list、存在性检查
• frappe.qb（query-builder 技能）——联接、GROUP BY、HAVING、子查询、跨数据库函数
• frappe.db.sql——无法通过 qb 表达的超复杂 SQL（始终参数化）

参考文件

• methods-reference.md——所有数据库和文档方法的完整 API 签名
• query-patterns.md——查询构建器模式、子查询、ImportMapper、自定义函数
• caching-patterns.md——Redis 缓存、@redis_cache、哈希操作、失效处理
• examples.md——真实场景模式：CRUD、报表、批处理、事务
• anti-patterns.md——SQL 注入、N+1、提交错误及另外 10 种反模式