应用可以通过三种方式扩展自动化流程：发出事件，让流程据此触发或等待；添加操作，加入步骤列表；以及添加关系，让步骤能够访问架构未直接关联的记录。每一种都在应用的 hooks.py 中注册。
发出事件
在应用中发生有意义的事情时，调用 emit()：
from frappe.automation_engine import emit
def on_deal_won(deal):
emit("crm.deal_won", doc=deal, payload={"amount": deal.deal_value})
| 参数 | 说明 |
|---|---|
event |
事件名称。使用带应用前缀的点分名称，例如 crm.deal_won。 |
doc |
事件所涉及的单据。带有文档类型的自定义事件流程会针对它运行。 |
payload |
一个可 JSON 序列化的字典。流程在模板中通过 payload 读取它，在条件中通过 context["payload"] 读取它。 |
correlation_key |
恢复暂停在匹配的 WaitForEvent 步骤上的运行。 |
emit() 做两件事：
- 它为该事件排队所有已启用的自定义事件流程。
- 如果提供了
correlation_key，它会恢复任何使用相同键等待该事件的运行。
它返回 {"queued": <count>, "resumed": <count>}</count></count>。运行会在当前事务提交后开始，在此之前对同一记录调用两次 emit() 只会产生一次运行。超过自动化设置中事件负载限制（默认 64 KB）的负载会被拒绝。
恢复等待中的流程
流程可以先发送一封邮件，然后用关联键 {{ doc.name }} 执行 WaitForEvent crm.email_replied。当回复到达时，你的应用用相同的键发出相同的事件：
emit(
"crm.email_replied",
doc=lead,
payload={"communication": comm.name},
correlation_key=lead.name,
)
等待中的运行会立即恢复，其后续步骤可以从 context["event"]["payload"] 读取负载。
注册事件
emit() 只接受已安装应用注册过的事件，除非在自动化设置中开启了允许未注册事件。在 hooks.py 中注册事件，可以是裸名称，也可以带有构建器界面可以显示的元数据：
automation_events = [
"crm.deal_won",
{
"crm.email_replied": {
"label": "Email replied",
"description": "A contact replied to an email sent from CRM",
"subject": {"doctype_key": "reference_doctype", "name_key": "reference_name", "doctypes": ["CRM Lead", "CRM Deal"]},
}
},
]
subject 告诉引擎事件涉及哪条记录，从负载中读取。有了主体，CRM Lead 上的流程会针对负载中指定的线索运行，即使发出事件的代码当时持有的是 Communication。当主体类型固定时，用 "doctype": "CRM Lead" 代替 doctype_key。
添加操作
继承 AutomationAction 并实现 execute：
# my_app/automation.py
from typing import ClassVar
import frappe
from frappe.automation_engine.actions.base import AutomationAction, AutomationParamError, render_value
class PostToChannel(AutomationAction):
action_type = "PostToChannel"
label = "Post to Channel"
description = "Post a message to a team channel."
transactional = False # it calls an external service
params_schema: ClassVar[list] = [
{"fieldname": "channel", "label": "Channel", "fieldtype": "Data", "reqd": 1},
{"fieldname": "message", "label": "Message", "fieldtype": "Small Text", "reqd": 1},
]
def validate(self, params, doctype):
if not params.get("channel"):
raise AutomationParamError("Channel is required", fieldname="channel")
def execute(self, doc, params, context):
message = render_value(params["message"], doc, context)
post_message(params["channel"], message)
return f"Posted to {params['channel']}"
# hooks.py
automation_actions = ["my_app.automation.PostToChannel"]
类属性：
| 属性 | 用途 |
|---|---|
action_type |
存储在步骤上的唯一标识符。发布后不要更改。 |
label、description |
显示在操作类型列表和构建器界面中。 |
params_schema |
参数字段：fieldname、label、fieldtype、options、reqd。 |
requires_document |
如果操作可以在没有触发单据的情况下运行（例如在计划流程上），则设为 False。 |
applicable_doctypes |
将操作限制为这些 DocType。None 表示全部。 |
supported_trigger_types |
将操作限制为这些触发类型。None 表示全部。 |
transactional |
如果操作具有数据库回滚无法撤销的副作用（例如 HTTP 调用或消息），则设为 False。这样引擎就永远不会从头重放该运行。 |
方法：
validate(params, doctype)：在流程保存时运行，并在每次执行前再次运行。抛出带有fieldname的AutomationParamError来指向错误的参数。execute(doc, params, context)：执行实际工作。返回一个短字符串用于运行日志，或返回一个包含detail键的字典，以及后续步骤应通过context["steps"][step_key]读取的任何输出。output_doctype(params)：如果操作创建了一条记录，返回其 DocType。这样步骤的输出别名就可以指向新记录，后续步骤也会针对它进行验证。
操作以流程的运行身份用户运行，因此正常的权限检查会生效。不要调用 frappe.db.commit()；事务由引擎管理。要从操作内部暂停运行，请抛出 StopAutomation("reason", resume_after=seconds)。
添加关系
Link、Dynamic Link 和子表链接字段会自动成为双向关系。只有当架构无法表达某种连接时（例如通过电子邮件地址匹配的记录），才需要注册提供程序：
from frappe.automation_engine.relationships import AutomationRelationshipProvider
class ContactRelationships(AutomationRelationshipProvider):
def get_definitions(self, source_doctype):
if source_doctype != "Communication":
return []
return [{
"name": "sender_lead",
"label": "Lead (by sender email)",
"target_doctype": "CRM Lead",
"cardinality": "one",
}]
def resolve(self, source_doc, relationship, params):
name = frappe.db.get_value("CRM Lead", {"email": source_doc.sender})
return [{"doctype": "CRM Lead", "name": name}] if name else []
# hooks.py
automation_relationships = ["my_app.automation.ContactRelationships"]
automation_relationship_ignore = ["My Log DocType"]
cardinality是one或many。只有one关系可以用作步骤的目标别名。many关系用于相关记录条件，此时覆盖query()以在 SQL 中过滤可以让它们在大表上保持高效。- 与派生关系具有相同
name的定义会替换它，这允许应用重新标记架构链接。 - 提供程序返回的每个引用在步骤使用之前都会检查其存在性和读取权限。
automation_relationship_ignore会从派生关系中隐藏 DocTypes（例如日志）。
抑制和检查自动化
from frappe.automation_engine import is_enabled, skip_automations
if is_enabled():
...
with skip_automations():
doc.save() # no Automation Flow fires for this save