server-scripts
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseFrappe Server Scripts Reference
Frappe服务器端脚本参考
Complete reference for server-side Python development in Frappe Framework.
Frappe框架下服务器端Python开发的完整参考文档。
When to Use This Skill
适用场景
- Writing document controllers
- Creating whitelisted API endpoints
- Handling document lifecycle events
- Background job processing
- Database operations and queries
- Permission checks and validation
- Email and notification handling
- 编写文档控制器
- 创建白名单API端点
- 处理文档生命周期事件
- 后台任务处理
- 数据库操作与查询
- 权限校验与数据验证
- 邮件与通知处理
Controller Location
控制器位置
my_app/
└── my_module/
└── doctype/
└── my_doctype/
└── my_doctype.py # Python controllermy_app/
└── my_module/
└── doctype/
└── my_doctype/
└── my_doctype.py # Python控制器Document Controller
文档控制器
Complete Controller Template
完整控制器模板
python
undefinedpython
undefinedmy_doctype.py
my_doctype.py
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import nowdate, nowtime, flt, cint, getdate, add_days
class MyDocType(Document):
# ===== NAMING =====
def autoname(self):
"""Custom naming logic"""
self.name = f"{self.prefix}-{frappe.generate_hash()[:8].upper()}"
def before_naming(self):
"""Called before autoname"""
pass
# ===== VALIDATION =====
def before_validate(self):
"""Called before validate"""
self.set_defaults()
def validate(self):
"""Main validation - called on insert and update"""
self.validate_dates()
self.validate_amounts()
self.calculate_totals()
self.set_status()
def before_save(self):
"""Called after validate, before database write"""
self.update_modified_info()
# ===== INSERT =====
def before_insert(self):
"""Called before new document is inserted"""
self.set_initial_values()
def after_insert(self):
"""Called after new document is inserted"""
self.create_related_documents()
self.send_notification()
# ===== UPDATE =====
def on_update(self):
"""Called after document is saved (insert or update)"""
self.update_related_documents()
self.clear_cache()
def after_save(self):
"""Called after on_update, always runs"""
pass
def on_change(self):
"""Called when document changes in database"""
pass
# ===== SUBMISSION =====
def before_submit(self):
"""Called before document is submitted"""
self.validate_for_submit()
def on_submit(self):
"""Called after document is submitted"""
self.create_gl_entries()
self.update_stock()
def on_update_after_submit(self):
"""Called when submitted doc is updated (limited fields)"""
pass
# ===== CANCELLATION =====
def before_cancel(self):
"""Called before document is cancelled"""
self.validate_cancellation()
def on_cancel(self):
"""Called after document is cancelled"""
self.reverse_gl_entries()
self.reverse_stock()
# ===== DELETION =====
def before_delete(self):
"""Called before document is deleted"""
self.check_dependencies()
def after_delete(self):
"""Called after document is deleted"""
self.cleanup_related()
def on_trash(self):
"""Called when document is trashed"""
pass
def after_restore(self):
"""Called after document is restored from trash"""
pass
# ===== CUSTOM METHODS =====
def set_defaults(self):
"""Set default values"""
if not self.posting_date:
self.posting_date = nowdate()
if not self.company:
self.company = frappe.defaults.get_user_default("Company")
def validate_dates(self):
"""Validate date fields"""
if self.end_date and getdate(self.start_date) > getdate(self.end_date):
frappe.throw(_("End Date cannot be before Start Date"))
if getdate(self.posting_date) > getdate(nowdate()):
frappe.throw(_("Posting Date cannot be in the future"))
def validate_amounts(self):
"""Validate amount fields"""
for item in self.items:
if flt(item.qty) <= 0:
frappe.throw(_("Row {0}: Quantity must be greater than 0").format(item.idx))
if flt(item.rate) < 0:
frappe.throw(_("Row {0}: Rate cannot be negative").format(item.idx))
def calculate_totals(self):
"""Calculate document totals"""
self.total = 0
for item in self.items:
item.amount = flt(item.qty) * flt(item.rate)
self.total += item.amount
self.tax_amount = flt(self.total) * flt(self.tax_rate) / 100
self.grand_total = flt(self.total) + flt(self.tax_amount)
def set_status(self):
"""Set document status based on state"""
if self.docstatus == 0:
self.status = "Draft"
elif self.docstatus == 1:
if self.is_completed():
self.status = "Completed"
else:
self.status = "Submitted"
elif self.docstatus == 2:
self.status = "Cancelled"
def is_completed(self):
"""Check if document is completed"""
return all(item.delivered_qty >= item.qty for item in self.items)undefinedimport frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import nowdate, nowtime, flt, cint, getdate, add_days
class MyDocType(Document):
# ===== 命名规则 =====
def autoname(self):
"""自定义命名逻辑"""
self.name = f"{self.prefix}-{frappe.generate_hash()[:8].upper()}"
def before_naming(self):
"""在autoname之前调用"""
pass
# ===== 数据验证 =====
def before_validate(self):
"""在validate之前调用"""
self.set_defaults()
def validate(self):
"""主验证逻辑 - 插入和更新时调用"""
self.validate_dates()
self.validate_amounts()
self.calculate_totals()
self.set_status()
def before_save(self):
"""验证完成后、写入数据库前调用"""
self.update_modified_info()
# ===== 插入操作 =====
def before_insert(self):
"""新文档插入前调用"""
self.set_initial_values()
def after_insert(self):
"""新文档插入后调用"""
self.create_related_documents()
self.send_notification()
# ===== 更新操作 =====
def on_update(self):
"""文档保存后调用(插入或更新)"""
self.update_related_documents()
self.clear_cache()
def after_save(self):
"""在on_update之后调用,始终执行"""
pass
def on_change(self):
"""文档在数据库中发生变化时调用"""
pass
# ===== 提交操作 =====
def before_submit(self):
"""文档提交前调用"""
self.validate_for_submit()
def on_submit(self):
"""文档提交后调用"""
self.create_gl_entries()
self.update_stock()
def on_update_after_submit(self):
"""已提交文档更新时调用(仅允许修改有限字段)"""
pass
# ===== 取消操作 =====
def before_cancel(self):
"""文档取消前调用"""
self.validate_cancellation()
def on_cancel(self):
"""文档取消后调用"""
self.reverse_gl_entries()
self.reverse_stock()
# ===== 删除操作 =====
def before_delete(self):
"""文档删除前调用"""
self.check_dependencies()
def after_delete(self):
"""文档删除后调用"""
self.cleanup_related()
def on_trash(self):
"""文档被移入回收站时调用"""
pass
def after_restore(self):
"""文档从回收站恢复后调用"""
pass
# ===== 自定义方法 =====
def set_defaults(self):
"""设置默认值"""
if not self.posting_date:
self.posting_date = nowdate()
if not self.company:
self.company = frappe.defaults.get_user_default("Company")
def validate_dates(self):
"""验证日期字段"""
if self.end_date and getdate(self.start_date) > getdate(self.end_date):
frappe.throw(_("结束日期不能早于开始日期"))
if getdate(self.posting_date) > getdate(nowdate()):
frappe.throw(_("过账日期不能是未来日期"))
def validate_amounts(self):
"""验证金额字段"""
for item in self.items:
if flt(item.qty) <= 0:
frappe.throw(_("第{0}行:数量必须大于0").format(item.idx))
if flt(item.rate) < 0:
frappe.throw(_("第{0}行:单价不能为负数").format(item.idx))
def calculate_totals(self):
"""计算文档总计"""
self.total = 0
for item in self.items:
item.amount = flt(item.qty) * flt(item.rate)
self.total += item.amount
self.tax_amount = flt(self.total) * flt(self.tax_rate) / 100
self.grand_total = flt(self.total) + flt(self.tax_amount)
def set_status(self):
"""根据状态设置文档状态"""
if self.docstatus == 0:
self.status = "草稿"
elif self.docstatus == 1:
if self.is_completed():
self.status = "已完成"
else:
self.status = "已提交"
elif self.docstatus == 2:
self.status = "已取消"
def is_completed(self):
"""检查文档是否已完成"""
return all(item.delivered_qty >= item.qty for item in self.items)undefinedWhitelisted APIs
白名单API
Basic API
基础API
python
@frappe.whitelist()
def get_customer_details(customer):
"""Get customer details
Args:
customer (str): Customer ID
Returns:
dict: Customer details with outstanding amount
"""
if not customer:
frappe.throw(_("Customer is required"))
doc = frappe.get_doc("Customer", customer)
return {
"customer_name": doc.customer_name,
"customer_type": doc.customer_type,
"territory": doc.territory,
"credit_limit": flt(doc.credit_limit),
"outstanding": get_customer_outstanding(customer)
}
@frappe.whitelist()
def create_invoice(customer, items):
"""Create sales invoice from data
Args:
customer (str): Customer ID
items (str): JSON string of items
Returns:
str: Invoice name
"""
items = frappe.parse_json(items)
doc = frappe.get_doc({
"doctype": "Sales Invoice",
"customer": customer,
"items": [{
"item_code": item.get("item_code"),
"qty": flt(item.get("qty")),
"rate": flt(item.get("rate"))
} for item in items]
})
doc.insert()
doc.submit()
return doc.namepython
@frappe.whitelist()
def get_customer_details(customer):
"""获取客户详情
参数:
customer (str): 客户ID
返回:
dict: 包含未结金额的客户详情
"""
if not customer:
frappe.throw(_("客户为必填项"))
doc = frappe.get_doc("Customer", customer)
return {
"customer_name": doc.customer_name,
"customer_type": doc.customer_type,
"territory": doc.territory,
"credit_limit": flt(doc.credit_limit),
"outstanding": get_customer_outstanding(customer)
}
@frappe.whitelist()
def create_invoice(customer, items):
"""根据数据创建销售发票
参数:
customer (str): 客户ID
items (str): 商品的JSON字符串
返回:
str: 发票名称
"""
items = frappe.parse_json(items)
doc = frappe.get_doc({
"doctype": "Sales Invoice",
"customer": customer,
"items": [{
"item_code": item.get("item_code"),
"qty": flt(item.get("qty")),
"rate": flt(item.get("rate"))
} for item in items]
})
doc.insert()
doc.submit()
return doc.nameGuest API
访客API
python
@frappe.whitelist(allow_guest=True)
def get_public_data():
"""Public API - no login required"""
return {
"status": "ok",
"message": "This is public data"
}python
@frappe.whitelist(allow_guest=True)
def get_public_data():
"""公开API - 无需登录"""
return {
"status": "ok",
"message": "这是公开数据"
}Method-Restricted API
方法限制API
python
@frappe.whitelist(methods=["POST"])
def create_record(data):
"""Only accepts POST requests"""
data = frappe.parse_json(data)
doc = frappe.get_doc(data)
doc.insert()
return {"name": doc.name}
@frappe.whitelist(methods=["GET", "POST"])
def flexible_endpoint(**kwargs):
"""Accepts GET and POST"""
return kwargspython
@frappe.whitelist(methods=["POST"])
def create_record(data):
"""仅接受POST请求"""
data = frappe.parse_json(data)
doc = frappe.get_doc(data)
doc.insert()
return {"name": doc.name}
@frappe.whitelist(methods=["GET", "POST"])
def flexible_endpoint(**kwargs):
"""接受GET和POST请求"""
return kwargsPermission-Checked API
权限校验API
python
@frappe.whitelist()
def sensitive_operation(doctype, name):
"""API with permission check"""
# Check permission
if not frappe.has_permission(doctype, "write", name):
frappe.throw(_("Not permitted"), frappe.PermissionError)
# Proceed with operation
doc = frappe.get_doc(doctype, name)
# ... do something
return {"status": "success"}python
@frappe.whitelist()
def sensitive_operation(doctype, name):
"""带权限校验的API"""
# 检查权限
if not frappe.has_permission(doctype, "write", name):
frappe.throw(_("无操作权限"), frappe.PermissionError)
# 执行操作
doc = frappe.get_doc(doctype, name)
# ... 执行操作
return {"status": "success"}Database Operations
数据库操作
Reading Data
读取数据
python
undefinedpython
undefinedGet single document
获取单个文档
doc = frappe.get_doc("Customer", "CUST-001")
doc = frappe.get_doc("Customer", "CUST-001")
Get with filters
带过滤条件获取
doc = frappe.get_doc("Customer", {"customer_name": "John Corp"})
doc = frappe.get_doc("Customer", {"customer_name": "John Corp"})
Get single value
获取单个字段值
name = frappe.db.get_value("Customer", "CUST-001", "customer_name")
name = frappe.db.get_value("Customer", "CUST-001", "customer_name")
Get multiple values
获取多个字段值
values = frappe.db.get_value("Customer", "CUST-001",
["customer_name", "territory"], as_dict=True)
values = frappe.db.get_value("Customer", "CUST-001",
["customer_name", "territory"], as_dict=True)
Get list
获取列表
customers = frappe.db.get_all("Customer",
filters={"status": "Active"},
fields=["name", "customer_name", "territory"],
order_by="customer_name asc",
limit=10
)
customers = frappe.db.get_all("Customer",
filters={"status": "Active"},
fields=["name", "customer_name", "territory"],
order_by="customer_name asc",
limit=10
)
Complex filters
复杂过滤条件
invoices = frappe.db.get_all("Sales Invoice",
filters={
"status": ["in", ["Paid", "Unpaid"]],
"grand_total": [">", 1000],
"posting_date": [">=", "2024-01-01"],
"customer": ["like", "%Corp%"]
},
fields=["name", "customer", "grand_total"]
)
invoices = frappe.db.get_all("Sales Invoice",
filters={
"status": ["in", ["Paid", "Unpaid"]],
"grand_total": [">", 1000],
"posting_date": [">=", "2024-01-01"],
"customer": ["like", "%Corp%"]
},
fields=["name", "customer", "grand_total"]
)
Pluck single field
提取单个字段列表
names = frappe.db.get_all("Customer",
filters={"status": "Active"},
pluck="name"
)
names = frappe.db.get_all("Customer",
filters={"status": "Active"},
pluck="name"
)
Count
计数
count = frappe.db.count("Customer", {"status": "Active"})
count = frappe.db.count("Customer", {"status": "Active"})
Exists check
存在性检查
exists = frappe.db.exists("Customer", "CUST-001")
undefinedexists = frappe.db.exists("Customer", "CUST-001")
undefinedRaw SQL
原生SQL
python
undefinedpython
undefinedSimple query
简单查询
result = frappe.db.sql("""
SELECT name, customer_name, grand_total
FROM
WHERE status = %s AND grand_total > %s
ORDER BY creation DESC
LIMIT 10
""", ("Paid", 1000), as_dict=True)
tabSales Invoiceresult = frappe.db.sql("""
SELECT name, customer_name, grand_total
FROM
WHERE status = %s AND grand_total > %s
ORDER BY creation DESC
LIMIT 10
""", ("Paid", 1000), as_dict=True)
tabSales InvoiceNamed parameters
命名参数查询
result = frappe.db.sql("""
SELECT * FROM
WHERE territory = %(territory)s
AND status = %(status)s
""", {"territory": "West", "status": "Active"}, as_dict=True)
tabCustomerresult = frappe.db.sql("""
SELECT * FROM
WHERE territory = %(territory)s
AND status = %(status)s
""", {"territory": "West", "status": "Active"}, as_dict=True)
tabCustomerAggregation
聚合查询
total = frappe.db.sql("""
SELECT SUM(grand_total) as total
FROM
WHERE status = 'Paid'
""")[0][0] or 0
tabSales Invoiceundefinedtotal = frappe.db.sql("""
SELECT SUM(grand_total) as total
FROM
WHERE status = 'Paid'
""")[0][0] or 0
tabSales InvoiceundefinedWriting Data
写入数据
python
undefinedpython
undefinedCreate document
创建文档
doc = frappe.get_doc({
"doctype": "Customer",
"customer_name": "New Customer",
"customer_type": "Company"
})
doc.insert()
doc = frappe.get_doc({
"doctype": "Customer",
"customer_name": "新客户",
"customer_type": "公司"
})
doc.insert()
Update document
更新文档
doc = frappe.get_doc("Customer", "CUST-001")
doc.customer_name = "Updated Name"
doc.save()
doc = frappe.get_doc("Customer", "CUST-001")
doc.customer_name = "更新后的名称"
doc.save()
Quick update (bypasses controller)
快速更新(绕过控制器)
frappe.db.set_value("Customer", "CUST-001", "status", "Inactive")
frappe.db.set_value("Customer", "CUST-001", "status", "Inactive")
Update multiple fields
更新多个字段
frappe.db.set_value("Customer", "CUST-001", {
"status": "Inactive",
"disabled": 1
})
frappe.db.set_value("Customer", "CUST-001", {
"status": "Inactive",
"disabled": 1
})
Delete
删除文档
frappe.delete_doc("Customer", "CUST-001")
frappe.delete_doc("Customer", "CUST-001")
Commit transaction
提交事务
frappe.db.commit()
frappe.db.commit()
Rollback
回滚事务
frappe.db.rollback()
undefinedfrappe.db.rollback()
undefinedBackground Jobs
后台任务
Enqueue Jobs
入队任务
python
undefinedpython
undefinedBasic enqueue
基础入队
frappe.enqueue(
"my_app.tasks.process_data",
queue="default",
customer="CUST-001"
)
frappe.enqueue(
"my_app.tasks.process_data",
queue="default",
customer="CUST-001"
)
With options
带选项的入队
frappe.enqueue(
method="my_app.tasks.heavy_task",
queue="long", # short, default, long
timeout=1800, # 30 minutes
is_async=True,
job_name="Heavy Task",
now=False, # True to run immediately
enqueue_after_commit=True,
# Task arguments
document_name="DOC-001",
data={"key": "value"}
)
undefinedfrappe.enqueue(
method="my_app.tasks.heavy_task",
queue="long", # short, default, long
timeout=1800, # 30分钟
is_async=True,
job_name="Heavy Task",
now=False, # True表示立即运行
enqueue_after_commit=True,
# 任务参数
document_name="DOC-001",
data={"key": "value"}
)
undefinedTask Function
任务函数
python
undefinedpython
undefinedmy_app/tasks.py
my_app/tasks.py
import frappe
def process_data(customer):
"""Background task"""
frappe.init(site=frappe.local.site)
frappe.connect()
try:
# Process logic
doc = frappe.get_doc("Customer", customer)
doc.last_processed = frappe.utils.now()
doc.save()
frappe.db.commit()
except Exception:
frappe.log_error(title="Process Data Failed")
raise
finally:
frappe.destroy()undefinedimport frappe
def process_data(customer):
"""后台任务"""
frappe.init(site=frappe.local.site)
frappe.connect()
try:
# 处理逻辑
doc = frappe.get_doc("Customer", customer)
doc.last_processed = frappe.utils.now()
doc.save()
frappe.db.commit()
except Exception:
frappe.log_error(title="数据处理失败")
raise
finally:
frappe.destroy()undefinedScheduled Jobs (hooks.py)
定时任务(hooks.py)
python
scheduler_events = {
"all": [
"my_app.tasks.every_minute"
],
"daily": [
"my_app.tasks.daily_report"
],
"hourly": [
"my_app.tasks.hourly_sync"
],
"cron": {
"0 9 * * 1": [
"my_app.tasks.monday_morning"
],
"*/15 * * * *": [
"my_app.tasks.every_15_min"
]
}
}python
scheduler_events = {
"all": [
"my_app.tasks.every_minute"
],
"daily": [
"my_app.tasks.daily_report"
],
"hourly": [
"my_app.tasks.hourly_sync"
],
"cron": {
"0 9 * * 1": [
"my_app.tasks.monday_morning"
],
"*/15 * * * *": [
"my_app.tasks.every_15_min"
]
}
}Error Handling
错误处理
python
from frappe import _
from frappe.exceptions import ValidationError, PermissionError
def my_function():
# Throw with message
frappe.throw(_("Invalid data"))
# Throw with title
frappe.throw(_("Cannot proceed"), title=_("Error"))
# Throw with exception type
frappe.throw(_("Permission denied"), exc=PermissionError)
# Message without stopping
frappe.msgprint(_("Warning: Check your data"))
# Log error
frappe.log_error(
title="My Error",
message=frappe.get_traceback()
)
# Try-except
try:
risky_operation()
except Exception:
frappe.log_error("Operation failed")
frappe.throw(_("Something went wrong"))python
from frappe import _
from frappe.exceptions import ValidationError, PermissionError
def my_function():
# 抛出错误信息
frappe.throw(_("数据无效"))
# 带标题的错误
frappe.throw(_("无法继续操作"), title=_("错误"))
# 指定异常类型抛出
frappe.throw(_("权限被拒绝"), exc=PermissionError)
# 仅提示不终止
frappe.msgprint(_("警告:请检查您的数据"))
# 记录错误日志
frappe.log_error(
title="自定义错误",
message=frappe.get_traceback()
)
# 异常捕获
try:
risky_operation()
except Exception:
frappe.log_error("操作失败")
frappe.throw(_("发生未知错误"))Utilities
工具函数
python
from frappe.utils import (
nowdate, nowtime, now_datetime, today,
getdate, get_datetime,
add_days, add_months, add_years,
date_diff, time_diff_in_seconds,
flt, cint, cstr,
fmt_money, rounded,
strip_html, escape_html
)python
from frappe.utils import (
nowdate, nowtime, now_datetime, today,
getdate, get_datetime,
add_days, add_months, add_years,
date_diff, time_diff_in_seconds,
flt, cint, cstr,
fmt_money, rounded,
strip_html, escape_html
)Date operations
日期操作
today = nowdate() # "2024-01-15"
week_later = add_days(nowdate(), 7)
month_end = frappe.utils.get_last_day(nowdate())
today = nowdate() # "2024-01-15"
week_later = add_days(nowdate(), 7)
month_end = frappe.utils.get_last_day(nowdate())
Number operations
数值操作
amount = flt(value, 2) # Float with precision
count = cint(value) # Integer
amount = flt(value, 2) # 带精度的浮点数
count = cint(value) # 整数转换
Formatting
格式化
money = fmt_money(1234.56, currency="USD")
money = fmt_money(1234.56, currency="USD")
Current user
当前用户
user = frappe.session.user
roles = frappe.get_roles()
undefineduser = frappe.session.user
roles = frappe.get_roles()
undefinedEmail & Notifications
通过Hooks配置文档事件
python
undefinedpython
undefinedSend email
hooks.py
frappe.sendmail(
recipients=["user@example.com"],
subject="Subject",
message="Email body",
reference_doctype="Sales Invoice",
reference_name="SINV-00001"
)
doc_events = {
"Sales Invoice": {
"validate": "my_app.overrides.validate_invoice",
"on_submit": "my_app.overrides.on_submit_invoice",
"on_cancel": "my_app.overrides.on_cancel_invoice"
},
"*": {
"on_update": "my_app.overrides.log_all_changes"
}
}
```pythonWith template
my_app/overrides.py
frappe.sendmail(
recipients=["user@example.com"],
subject="Order Confirmation",
template="order_confirmation",
args={
"customer_name": "John",
"order_id": "ORD-001"
}
)
import frappe
def validate_invoice(doc, method):
"""销售发票验证时调用"""
if doc.grand_total > 100000:
if not doc.manager_approval:
frappe.throw(_("订单金额超过100,000需要经理审批"))
def on_submit_invoice(doc, method):
"""销售发票提交时调用"""
create_delivery_note(doc)
notify_warehouse(doc)
undefinedReal-time notification
—
frappe.publish_realtime(
"msgprint",
{"message": "Task completed"},
user="user@example.com"
)
undefined—
Document Events via Hooks
—
python
undefined—
hooks.py
—
doc_events = {
"Sales Invoice": {
"validate": "my_app.overrides.validate_invoice",
"on_submit": "my_app.overrides.on_submit_invoice",
"on_cancel": "my_app.overrides.on_cancel_invoice"
},
"*": {
"on_update": "my_app.overrides.log_all_changes"
}
}
```python—
my_app/overrides.py
—
import frappe
def validate_invoice(doc, method):
"""Called during Sales Invoice validation"""
if doc.grand_total > 100000:
if not doc.manager_approval:
frappe.throw(_("Manager approval required for orders above 100,000"))
def on_submit_invoice(doc, method):
"""Called when Sales Invoice is submitted"""
create_delivery_note(doc)
notify_warehouse(doc)
undefined—