[["status", "in", ["Open", "Pending"]]]not inNOT IN{"role": ["not in", ["Guest"]]}[["role", "not in", ["Guest"]]]isIS NULLIS NOT NULL{"customer": ["is", "set"]}{"email": ["is", "not set"]}[["customer", "is", "set"]][["email", "is", "not set"]]descendants of{"parent_account": ["descendants of", "Assets"]}[["parent_account", "descendants of", "Assets"]]ancestors of{"location": ["ancestors of", "Room 101"]}[["location", "ancestors of", "Room 101"]]not descendants of{"category": ["not descendants of", "Internal"]}[["category", "not descendants of", "Internal"]]not ancestors of{"territory": ["not ancestors of", "West Coast"]}[["territory", "not ancestors of", "West Coast"]]is setis not setIS NOT NULLIS NULLlink_fieldname.target_fieldnamechild_table_fieldname.target_fieldnamedistinct=Truelftrgtfilters'and''or'query.run()as_iterator=Trueas_iterator=Trueas_dict=Trueas_list=Truefrappe.db.unbuffered_cursor()order_bygroup_bylimitoffsetdistinct=Trueignore_permissionsfrappe.qb.get_queryignore_permissions=Trueignore_permissions=Falseignore_permissions=Falseif_ownerfieldsfiltersgroup_byorder_bylink_field.target_fieldchild_field.target_fieldignore_permissions=Falsefieldsfiltersfrappe.PermissionErrorgroup_byfrappe.PermissionErrororder_byfrappe.PermissionErrorlink_field.target_fieldchild_field.target_fieldfrappe.qb.get_queryignore_permissions=False[["status", "in", ["Open", "Pending"]]]not inNOT IN{"role": ["not in", ["Guest"]]}[["role", "not in", ["Guest"]]]isIS NULL 或 IS NOT NULL{"customer": ["is", "set"]} 或 {"email": ["is", "not set"]}[["customer", "is", "set"]] 或 [["email", "is", "not set"]]descendants of{"parent_account": ["descendants of", "Assets"]}[["parent_account", "descendants of", "Assets"]]ancestors of{"location": ["ancestors of", "Room 101"]}[["location", "ancestors of", "Room 101"]]not descendants of{"category": ["not descendants of", "Internal"]}[["category", "not descendants of", "Internal"]]not ancestors of{"territory": ["not ancestors of", "West Coast"]}[["territory", "not ancestors of", "West Coast"]]
关于 is set / is not set 的说明： 这些用于检查字段是否有值（分别为 IS NOT NULL 和 IS NULL）。
按链接文档字段筛选
您可以使用点号表示法根据链接文档中的字段进行筛选：link_fieldname.target_fieldname。
# Get Sales Orders where the linked Customer's territory is 'North America'
query = frappe.qb.get_query(
"Sales Order",
fields=["name", "customer"],
filters={"customer.territory": "North America"} # Filter on linked field
)
north_america_orders = query.run(as_dict=True)
按子表字段筛选
您可以使用点号表示法根据子表记录中的值筛选父记录：child_table_fieldname.target_fieldname。
# Get Sales Orders that contain 'Item A' in their items table
# Use distinct=True to ensure each Sales Order appears only once
query = frappe.qb.get_query(
"Sales Order",
fields=["name", "customer"],
filters={"items.item_code": "Item A"}, # Filter based on child table field
distinct=True
)
orders_with_item_a = query.run(as_dict=True)
重要提示： 当基于子表字段进行筛选时，如果您只需要唯一的父记录，请使用 distinct=True。
嵌套集筛选
对于树形结构的 DocType（使用 lft 和 rgt 列，如科目、地区、仓库等），您可以使用特殊筛选器：
# Get all accounts under 'Assets'
query = frappe.qb.get_query(
"Account",
fields=["name"],
filters={"parent_account": ["descendants of", "Assets"]}
)
# Get the parent territories of 'West Coast'
query = frappe.qb.get_query(
"Territory",
fields=["name"],
filters={"parent_territory": ["ancestors of", "West Coast"]}
)
逻辑运算符（AND/OR）
对于复杂条件，请将您的 filters 构建为列表，并使用 'and' 或 'or' 组合条件。
# Find users who are enabled AND have first name 'Admin'
filters_and = [
["enabled", "=", 1],
"and",
["first_name", "=", "Admin"],
]
query = frappe.qb.get_query("User", filters=filters_and)
# Find users who have first name 'Admin' OR 'Guest'
filters_or = [
["first_name", "=", "Admin"],
"or",
["first_name", "=", "Guest"],
]
query = frappe.qb.get_query("User", filters=filters_or)
# Combine AND and OR (use nested lists for grouping)
# Find users who are enabled AND (have first name 'Admin' OR 'Guest')
filters_nested = [
["enabled", "=", 1],
"and",
[
["first_name", "=", "Admin"],
"or",
["first_name", "=", "Guest"],
]
]
query = frappe.qb.get_query("User", filters=filters_nested)
查询执行
基本执行
一旦您获得了 query 对象，请使用 .run() 执行它：
# Returns a list of tuples by default
results = query.run()
# Returns a list of dictionaries
results = query.run(as_dict=True)
# Returns a list of lists
results = query.run(as_list=True)
# If selecting a single field, returns a flat list of values
results = query.run(pluck=True)
# Print the generated SQL query and execution time
results = query.run(debug=True)
获取 SQL 字符串
您可以在不执行的情况下获取生成的 SQL 字符串：
# Get the SQL string with values directly substituted (for debugging)
sql_string = query.get_sql()
print(sql_string)
# Example Output: SELECT `name`, `email` FROM `tabUser` WHERE `first_name`='Admin'
对大型数据集使用迭代器
处理大型数据集时，请使用 as_iterator=True 逐行处理结果，而无需将所有内容加载到内存中：
# Process a large number of tasks without loading all into memory
query = frappe.qb.get_query(
"Task",
fields=["name", "subject", "status"],
filters={"status": "Open"}
)
# Use unbuffered_cursor for optimal memory usage with the iterator
with frappe.db.unbuffered_cursor():
task_iterator = query.run(as_iterator=True, as_dict=True)
processed_count = 0
for task in task_iterator:
# Process each task dictionary one by one
print(f"Processing Task: {task['name']} - {task['subject']}")
processed_count += 1
if processed_count % 1000 == 0:
print(f"Processed {processed_count} tasks...")
要求：
- 您必须将
as_iterator=True与as_dict=True或as_list=True一起使用 - 为获得最佳内存效率，请在
frappe.db.unbuffered_cursor()上下文管理器中使用
排序、分组和分页
排序结果
使用 order_by 参数对结果进行排序：
# Order users by creation date, ascending
query = frappe.qb.get_query("User", fields=["name", "creation"], order_by="creation asc")
# Order by multiple fields
query = frappe.qb.get_query(
"Sales Invoice",
fields=["name", "customer", "grand_total"],
order_by="customer asc, grand_total desc"
)
分组结果
使用 group_by 进行聚合：
# Count invoices per customer
query = frappe.qb.get_query(
"Sales Invoice",
fields=["customer", {"COUNT": "'*'", "as": "invoice_count"}],
filters={"docstatus": 1},
group_by="customer"
)
results = query.run(as_dict=True)
# results: [{'customer': 'Cust A', 'invoice_count': 5}, {'customer': 'Cust B', 'invoice_count': 3}, ...]
分页
使用 limit 和 offset 进行分页：
# Get the first 10 users
query = frappe.qb.get_query("User", limit=10)
# Get the next 10 users (page 2)
query = frappe.qb.get_query("User", limit=10, offset=10)
去重结果
使用 distinct=True 获取唯一行：
# Get distinct customers from submitted Sales Invoices
query = frappe.qb.get_query(
"Sales Invoice",
fields=["customer"],
filters={"docstatus": 1},
distinct=True
)
权限
ignore_permissions 标志
默认情况下，frappe.qb.get_query 忽略权限（ignore_permissions=True）。要强制执行权限，请设置 ignore_permissions=False：
# This query bypasses all permission checks (default behavior)
query_ignore = frappe.qb.get_query("DocType", fields=["name"], filters={"istable": 1})
# This query enforces standard Frappe permissions for the current user
query_enforce = frappe.qb.get_query(
"DocType",
fields=["name"],
filters={"istable": 1},
ignore_permissions=False # Explicitly enable permission checks
)
try:
results = query_enforce.run()
except frappe.PermissionError:
print("User does not have permission to read DocType!")
权限的应用方式
当 ignore_permissions=False 时：
- 角色权限： 根据用户的角色检查其是否具有“读取”或“选择”权限。
- 用户权限： 应用为 DocType 和链接的 DocType 定义的用户权限（允许/限制）。
- 共享： 包含明确共享给用户的文档。
- 所有者约束： 如果角色权限仅授予
if_owner访问权限，则查询会将结果限制为用户拥有的文档。 - 权限查询条件： 应用通过 Hooks 或服务器脚本定义的条件。
- 字段级安全： 筛选所选的
fields，如果用户没有权限级别访问权限，则不允许使用filters、group_by、order_by中使用的字段。同时检查link_field.target_field和child_field.target_field表示法中的字段。
字段级安全
当 ignore_permissions=False 时：
fields： 仅包含用户最大允许权限级别下可访问的字段。请求不可访问的字段将静默移除该字段的选择。filters： 筛选仅允许在用户有权访问的字段上进行。尝试筛选不可访问的字段将引发frappe.PermissionError。group_by： 分组仅允许在用户有权访问的字段上进行。尝试按不可访问的字段分组将引发frappe.PermissionError。order_by： 仅允许对用户有权限访问的字段进行排序。尝试按无权限访问的字段排序将引发frappe.PermissionError。- 链接表和子表字段： 在上述任何子句中使用
link_field.target_field或child_field.target_field表示法时，系统会同时检查链接/子字段本身的权限，以及链接/子 DocType 中目标字段的权限。
# Assume 'published' field in Blog Post has permlevel 1
# User '[email protected]' only has permlevel 0 access
# This works, but 'published' field is silently removed from results
query = frappe.qb.get_query(
"Blog Post",
fields=["name", "title", "published"], # 'published' requested but inaccessible
ignore_permissions=False,
user="[email protected]"
)
# result will contain 'name' and 'title', but NOT 'published'
# This FAILS because filtering on 'published' is not allowed for this user
try:
query = frappe.qb.get_query(
"Blog Post",
fields=["name"],
filters={"published": 1}, # Filtering on restricted field
ignore_permissions=False,
user="[email protected]"
)
query.run()
except frappe.PermissionError as e:
print(f"Permission error: {e}")
指定用户和父级上下文
# Check permissions for a specific user
query = frappe.qb.get_query(
"Task",
ignore_permissions=False,
user="[email protected]" # Check permissions for this user
)
# Provide parent context for child DocTypes
query = frappe.qb.get_query(
"Sales Order Item",
fields=["item_code", "qty"],
filters={"parent": "SO-00001"},
ignore_permissions=False,
parent_doctype="Sales Order" # Specify parent context for permission checks
)
高级功能
使用 Pypika 对象
对于涉及子查询、高级条件或字典语法中不可用函数的复杂场景，您可以直接使用 Pypika 对象。
字段中的 Pypika 对象
from frappe.query_builder import Field, functions, Query
# Define Pypika objects
user_table = frappe.qb.DocType("User")
todo_table = frappe.qb.DocType("ToDo")
# Build a subquery to count open ToDos for each user
open_todo_subquery = (
Query.from_(todo_table)
.select(functions.Count("*"))
.where(todo_table.owner == user_table.name) # Correlated subquery
.where(todo_table.status == "Open")
).as_("open_todos_count")
# Use Pypika objects in fields
query = frappe.qb.get_query(
"User",
fields=[
user_table.name,
user_table.email,
open_todo_subquery # Using the subquery object
],
filters={"user_type": "System User"}
)
users_with_counts = query.run(as_dict=True)
过滤器中的 Pypika 对象
from frappe.query_builder import Field, functions
# Define Pypika objects
task_table = frappe.qb.DocType("Task")
modified_field = task_table.modified
creation_field = task_table.creation
subject_field = task_table.subject
status_field = task_table.status
# Build complex criterion
complex_filter = (
(modified_field > creation_field) & (functions.Length(subject_field) > 10)
) | (status_field == "Cancelled")
# Use the Criterion object in filters
query = frappe.qb.get_query(
"Task",
fields=["name", "subject", "status", "creation", "modified"],
filters=complex_filter
)
results = query.run(as_dict=True)
记录锁定
用于数据库事务中，防止其他事务修改特定行：
基本锁定
# Lock specific Stock Ledger Entries
query = frappe.qb.get_query(
"Stock Ledger Entry",
fields=["name", "qty_after_transaction"],
filters={"item_code": "ITEM001", "warehouse": "WH001"},
for_update=True # Adds FOR UPDATE clause, will wait if rows are locked
)
entries = query.run(as_dict=True)
跳过已锁定行
# Skip rows that are already locked by another transaction
query = frappe.qb.get_query(
"ToDo",
fields=["name", "description"],
filters={"status": "Pending"},
limit=5,
order_by="creation asc",
for_update=True,
skip_locked=True # Skip locked rows
)
available_tasks = query.run(as_dict=True)
非阻塞锁定尝试
# Fail immediately if rows are already locked
try:
query = frappe.qb.get_query(
"System Settings",
fields=["name"],
filters={"name": "System Settings"},
for_update=True,
wait=False # Don't wait for locks
)
settings = query.run(as_dict=True)
except Exception as e:
print(f"Could not acquire lock immediately: {e}")
安全注意事项
frappe.qb.get_query 在设计时充分考虑了安全性：
- 字段验证： 字段名称会按照严格的模式进行验证，以防止 SQL 注入。
- 参数化： 过滤器值由数据库驱动程序进行参数化处理。
- 权限执行： 使用
ignore_permissions=False可充分利用 Frappe 强大的权限系统。
请始终确保，如果用于构建过滤器键或字段名称的任何动态值来自不受信任的来源，都经过适当的清理。始终依赖将用户输入作为过滤器值传递。
由 Claude Sonnet 4 编写。经人工审核。