feat: App-Defined Workflow Tasks and Server Scripts on Transition (#33255)

* feat: add workflow task doctypes

* chore: add Workflow Task to Script Type of Server Scripts

* chore: add description for Workflow Tasks

* feat: fetch dotted paths and corresponding names from hooks

* fix: use Select field instead of Autocomplete

* feat: execute tasks in the background on state transition

* fix: throw error when task not found

* fix: naming for transition task set by user

* fix: child table views

* feat: support for server scripts

* chore: comments and description

* fix: change the casing of docevent options

* fix: change test casing

* feat: Add Workflow Transition field to webhooks

* fix: temporarily remove filter

* feat: add webhooks as workflow actions

* test: test execution of synchronous app-defined methods and server scripts

* chore: shorten the description and change the fieldname of the field 'execute_asynchronously' to 'asynchronous'

* test: server script executipn

* fix: add field to UI

* test: change customer to domain

* test: change Customer to Domain

* fix: patch for change in webhook fields

* chore: fetch only used fields in get_all

* fix: don't run test logic in application code 🙂

* test: separate tests for workflow tasks

* Revert "fix: patch for change in webhook fields"

This reverts commit 2e9c51c43ca2b3698991fbe75cc4032368ab174c.

* chore: break into smaller functions

* test: webhooks with workflow tasks

isolate mock responses to test_sync_tasks

* Revert "fix: change test casing"

This reverts commit e2bad96ed01dea0a26d002ea44a1e0175525ed31.

* Revert "fix: change the casing of docevent options"

This reverts commit ddfc81bf775fad74225f25815b8e3e7b255dc9eb.

* fix: webhook casing

* fix: type hint for doc
This commit is contained in:
Ayush Chaudhari 2025-07-29 11:01:44 +05:30 committed by GitHub
parent 6d1008933f
commit 761751f269
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 496 additions and 9 deletions

View file

@ -129,6 +129,21 @@ select name from \`tabPerson\`
where tenant_id = 2
order by creation desc
</code></pre>
<hr>
<h4>Workflow Task</h4>
<p>Execute when a particular <a href="/app/workflow-action-master">Workflow Action Master</a> is executed.</p>
<p>Gets the document which the action is being applied on in the <code>doc</code> variable.</p>
<code><pre>
# create a customer with the same name as the given document
customer = frappe.new_doc("Customer")
customer.customer_name = doc.first_name + " " + doc.last_name # we get this from the workflow action
customer.customer_type = "Company"
c.save()
</code></pre>
`);
},
});

View file

@ -32,7 +32,7 @@
"in_list_view": 1,
"in_standard_filter": 1,
"label": "Script Type",
"options": "DocType Event\nScheduler Event\nPermission Query\nAPI",
"options": "DocType Event\nScheduler Event\nPermission Query\nAPI\nWorkflow Task",
"reqd": 1
},
{
@ -151,7 +151,7 @@
"link_fieldname": "server_script"
}
],
"modified": "2024-05-08 03:21:54.169380",
"modified": "2025-07-03 16:12:29.676150",
"modified_by": "Administrator",
"module": "Core",
"name": "Server Script",
@ -171,6 +171,7 @@
"write": 1
}
],
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": [],

View file

@ -80,7 +80,9 @@ class ServerScript(Document):
rate_limit_seconds: DF.Int
reference_doctype: DF.Link | None
script: DF.Code
script_type: DF.Literal["DocType Event", "Scheduler Event", "Permission Query", "API"]
script_type: DF.Literal[
"DocType Event", "Scheduler Event", "Permission Query", "API", "Workflow Task"
]
# end: auto-generated types
def validate(self):
@ -216,6 +218,19 @@ class ServerScript(Document):
if locals["conditions"]:
return locals["conditions"]
def execute_workflow_task(self, doc: Document):
"""
Specific to Workflow Tasks via Workflow Action Master
"""
if self.script_type != "Workflow Task":
raise frappe.DoesNotExistError
safe_exec(
self.script,
_locals={"doc": doc},
script_filename=self.name,
)
@frappe.whitelist()
@http_cache(max_age=10 * 60, stale_while_revalidate=6 * 60 * 60)

View file

@ -56,7 +56,7 @@
"fieldtype": "Select",
"in_list_view": 1,
"label": "Doc Event",
"options": "after_insert\non_update\non_submit\non_cancel\non_trash\non_update_after_submit\non_change",
"options": "after_insert\non_update\non_submit\non_cancel\non_trash\non_update_after_submit\non_change\nworkflow_transition",
"set_only_once": 1
},
{
@ -189,7 +189,7 @@
"link_fieldname": "webhook"
}
],
"modified": "2024-10-28 12:21:52.172428",
"modified": "2025-07-18 18:22:38.276809",
"modified_by": "Administrator",
"module": "Integrations",
"name": "Webhook",

View file

@ -49,6 +49,7 @@ class Webhook(Document):
"on_trash",
"on_update_after_submit",
"on_change",
"workflow_transition",
]
webhook_doctype: DF.Link
webhook_headers: DF.Table[WebhookHeader]
@ -68,6 +69,9 @@ class Webhook(Document):
def on_update(self):
frappe.client_cache.delete_value("webhooks")
def execute_for_doc(self, doc: Document):
enqueue_webhook(doc, self)
def validate_docevent(self):
if self.webhook_doctype:
is_submittable = frappe.get_value("DocType", self.webhook_doctype, "is_submittable")
@ -144,7 +148,8 @@ def get_context(doc):
def enqueue_webhook(doc, webhook) -> None:
request_url = headers = data = r = None
try:
webhook: Webhook = frappe.get_doc("Webhook", webhook.get("name"))
if not isinstance(webhook, Document):
webhook: Webhook = frappe.get_doc("Webhook", webhook.get("name"))
request_url = webhook.request_url
if webhook.is_dynamic_url:
request_url = frappe.render_template(webhook.request_url, get_context(doc))
@ -180,6 +185,9 @@ def enqueue_webhook(doc, webhook) -> None:
sleep(3 * i + 1)
if i != 2:
continue
else:
if webhook.webhook_docevent == "workflow_transition":
raise e
def log_request(

View file

@ -14,6 +14,9 @@ if TYPE_CHECKING:
from frappe.workflow.doctype.workflow.workflow import Workflow
DEFAULT_WORKFLOW_TASKS = ["Webhook", "Server Script"]
class WorkflowStateError(frappe.ValidationError):
pass
@ -126,6 +129,59 @@ def apply_workflow(doc, action):
if next_state.update_field:
doc.set(next_state.update_field, next_state.update_value)
if transition.transition_tasks:
workflow_transitions = frappe.db.get_all(
"Workflow Transition Task",
{"parent": transition.transition_tasks, "enabled": True},
["task", "link", "asynchronous"],
order_by="idx",
)
"""app-specific actions defined by the user
Example:
def create_customer(doc):
<your-code>
this goes in the hooks.py
workflow_methods = [{"name": "Create a customer", "method":
"frappe.dotted.path.create_customer"}]
"""
tasks = {i["name"]: i["method"] for i in frappe.get_hooks("workflow_methods")}
sync_tasks = []
async_tasks = []
for workflow_transition in workflow_transitions:
# edge-case with user-defined server scripts
if workflow_transition.task in DEFAULT_WORKFLOW_TASKS:
match workflow_transition.task:
case "Webhook":
webhook = frappe.get_doc("Webhook", workflow_transition.link)
task_method = webhook.execute_for_doc
case "Server Script":
server_script = frappe.get_doc("Server Script", workflow_transition.link)
task_method = server_script.execute_workflow_task
else: # normal app-defined tasks
try:
task_method = frappe.get_attr(tasks[workflow_transition.task])
except KeyError:
frappe.throw(_('There is no task called "{}"').format(workflow_transition.task))
if workflow_transition.asynchronous:
async_tasks.append(task_method)
else:
sync_tasks.append(task_method)
# will execute in the same transaction as the rest of the transition
for sync_task in sync_tasks:
sync_task(doc)
# will spawn separate background jobs. Use for asynchronous, optional tasks.
for async_task in async_tasks:
frappe.enqueue(async_task, doc=doc, enqueue_after_commit=True)
new_docstatus = DocStatus(next_state.doc_status or 0)
if doc.docstatus.is_draft() and new_docstatus.is_draft():
doc.save()

View file

@ -18,7 +18,9 @@ let properties = computed(() => {
if (store.workflow.selected && "action" in store.workflow.selected.data) {
title.value = __("Transition Properties");
return store.transitionfields.filter((df) =>
["action", "allowed", "allow_self_approval", "condition"].includes(df.fieldname)
["action", "allowed", "allow_self_approval", "condition", "transition_tasks"].includes(
df.fieldname
)
);
} else if (store.workflow.selected && "state" in store.workflow.selected.data) {
title.value = __("State Properties");

View file

@ -2,13 +2,14 @@
# License: MIT. See LICENSE
from unittest.mock import patch
import responses
import frappe
from frappe.model.workflow import (
WorkflowTransitionError,
apply_workflow,
get_common_transition_actions,
)
from frappe.query_builder import DocType
from frappe.tests import IntegrationTestCase
from frappe.tests.utils import make_test_records
from frappe.utils import random_string
@ -19,16 +20,19 @@ class TestWorkflow(IntegrationTestCase):
def setUpClass(cls):
super().setUpClass()
make_test_records("User")
cls.enterClassContext(cls.enable_safe_exec())
def setUp(self):
self.patcher = patch("frappe.attach_print", return_value={})
self.patcher.start()
frappe.db.delete("Workflow Action")
self.workflow = create_todo_workflow()
create_domain_workflow()
def tearDown(self):
frappe.set_user("Administrator")
self.patcher.stop()
frappe.delete_doc("Workflow", "Test ToDo")
def test_default_condition(self):
@ -126,6 +130,50 @@ class TestWorkflow(IntegrationTestCase):
"invalid python code" in str(se.exception).lower(), msg="Python code validation not working"
)
# app-defined workflow task tests start here
def test_sync_tasks(self, doc=None):
"""test workflow with workflow tasks (server scripts, webhooks and app-defined methods)"""
# for webhooks
self.responses = responses.RequestsMock()
self.responses.start()
self.responses.add(
responses.POST,
"https://workflowtasks.org/post",
status=200,
json={},
)
domain = frappe.new_doc("Domain")
domain.domain = random_string(length=10)
domain.save()
with self.patch_hooks(
{
"workflow_methods": [
{
"name": "Create Note",
"method": "frappe.workflow.doctype.workflow.test_workflow.create_new_note",
}
]
}
):
apply_workflow(domain, "Approve")
# refer create_new_task()
self.assertTrue(
frappe.db.exists("Note", {"title": "workflow - " + domain.name, "content": "workflow test"})
)
self.assertTrue(frappe.db.exists("Domain", {"name": "workflow - " + domain.name}))
self.assertTrue(frappe.db.exists("Webhook Request Log", {"url": "https://workflowtasks.org/post"}))
# for webhooks
self.responses.stop()
self.responses.reset()
return domain
def create_todo_workflow():
from frappe.tests.ui_test_helpers import UI_TEST_USER
@ -181,5 +229,107 @@ def create_todo_workflow():
return workflow
def create_domain_workflow():
from frappe.tests.ui_test_helpers import UI_TEST_USER
if frappe.db.exists("Workflow", "Test Domain"):
frappe.delete_doc("Workflow", "Test Domain")
TEST_ROLE = "Test Approver"
if not frappe.db.exists("Role", TEST_ROLE):
frappe.get_doc(doctype="Role", role_name=TEST_ROLE).insert(ignore_if_duplicate=True)
if frappe.db.exists("User", UI_TEST_USER):
frappe.get_doc("User", UI_TEST_USER).add_roles(TEST_ROLE)
server_script = create_new_server_script()
webhook = create_new_webhook()
pending_to_approved_transition = frappe.new_doc("Workflow Transition Tasks")
pending_to_approved_transition.name = random_string(length=10)
pending_to_approved_transition.append("tasks", {"task": "Create Note"})
pending_to_approved_transition.append("tasks", {"task": "Server Script", "link": server_script.name})
pending_to_approved_transition.append("tasks", {"task": "Webhook", "link": webhook.name})
pending_to_approved_transition.save()
workflow = frappe.new_doc("Workflow")
workflow.workflow_name = "Test Domain"
workflow.document_type = "Domain"
workflow.workflow_state_field = "workflow_state"
workflow.is_active = 1
workflow.send_email_alert = 1
workflow.append("states", dict(state="Pending", allow_edit="All"))
workflow.append(
"states",
dict(state="Approved", allow_edit=TEST_ROLE, update_field="status", update_value="Closed"),
)
workflow.append("states", dict(state="Rejected", allow_edit=TEST_ROLE))
workflow.append(
"transitions",
dict(
state="Pending",
action="Approve",
next_state="Approved",
allowed=TEST_ROLE,
allow_self_approval=1,
transition_tasks=pending_to_approved_transition.name,
),
)
workflow.append(
"transitions",
dict(
state="Pending",
action="Reject",
next_state="Rejected",
allowed=TEST_ROLE,
allow_self_approval=1,
),
)
workflow.append(
"transitions",
dict(state="Rejected", action="Review", next_state="Pending", allowed="All", allow_self_approval=1),
)
workflow.insert(ignore_permissions=True)
return workflow
def create_new_todo():
return frappe.get_doc(doctype="ToDo", description="workflow " + random_string(10)).insert()
def create_new_note(doc):
note = frappe.new_doc("Note")
note.title = "workflow - " + doc.name
note.content = "workflow test"
note.save()
def create_new_server_script():
server_script = frappe.new_doc("Server Script")
server_script.name = random_string(length=10)
server_script.script_type = "Workflow Task"
server_script.script = """
# create a domain with the same name as the given document
domain = frappe.new_doc("Domain")
domain.domain = "workflow - " + doc.name
domain.save()
"""
server_script.save()
return server_script
def create_new_webhook():
webhook = frappe.new_doc("Webhook")
webhook.__newname = random_string(10)
webhook.webhook_docevent = "workflow_transition"
webhook.webhook_doctype = "Domain"
webhook.request_method = "POST"
webhook.request_url = "https://workflowtasks.org/post"
webhook.save()
return webhook

View file

@ -4,6 +4,7 @@
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.model.workflow import DEFAULT_WORKFLOW_TASKS
from frappe.utils import cint
@ -131,3 +132,8 @@ def get_workflow_state_count(doctype, workflow_state_field, states):
group_by=workflow_state_field,
)
return [r for r in result if r[workflow_state_field]]
@frappe.whitelist(methods=["GET"])
def get_workflow_methods():
return [i["name"] for i in frappe.get_hooks("workflow_methods")] + DEFAULT_WORKFLOW_TASKS

View file

@ -12,6 +12,7 @@
"allowed",
"allow_self_approval",
"send_email_to_creator",
"transition_tasks",
"conditions",
"condition",
"column_break_7",
@ -99,17 +100,24 @@
"fieldname": "send_email_to_creator",
"fieldtype": "Check",
"label": "Send Email To Creator"
},
{
"fieldname": "transition_tasks",
"fieldtype": "Link",
"label": "Transition Tasks",
"options": "Workflow Transition Tasks"
}
],
"idx": 1,
"istable": 1,
"links": [],
"modified": "2025-03-24 02:47:44.188152",
"modified": "2025-07-04 15:56:34.345888",
"modified_by": "Administrator",
"module": "Workflow",
"name": "Workflow Transition",
"owner": "Administrator",
"permissions": [],
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": []

View file

@ -24,6 +24,7 @@ class WorkflowTransition(Document):
parenttype: DF.Data
send_email_to_creator: DF.Check
state: DF.Link
transition_tasks: DF.Link | None
workflow_builder_id: DF.Data | None
# end: auto-generated types

View file

@ -0,0 +1,20 @@
# Copyright (c) 2025, Frappe Technologies and Contributors
# See license.txt
# import frappe
from frappe.tests import IntegrationTestCase
# On IntegrationTestCase, the doctype test records and all
# link-field test record dependencies are recursively loaded
# Use these module variables to add/remove to/from that list
EXTRA_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
IGNORE_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
class IntegrationTestWorkflowTransitionTask(IntegrationTestCase):
"""
Integration tests for WorkflowTransitionTask.
Use this class for testing interactions between multiple components.
"""
pass

View file

@ -0,0 +1,8 @@
// Copyright (c) 2025, Frappe Technologies and contributors
// For license information, please see license.txt
// frappe.ui.form.on("Workflow Transition Task", {
// refresh(frm) {
// },
// });

View file

@ -0,0 +1,63 @@
{
"actions": [],
"allow_rename": 1,
"creation": "2025-07-04 16:41:15.904217",
"doctype": "DocType",
"engine": "InnoDB",
"field_order": [
"task",
"asynchronous",
"enabled",
"link"
],
"fields": [
{
"fieldname": "task",
"fieldtype": "Select",
"in_list_view": 1,
"in_preview": 1,
"label": "Task",
"reqd": 1
},
{
"default": "1",
"fieldname": "enabled",
"fieldtype": "Check",
"in_list_view": 1,
"in_preview": 1,
"label": "Enabled"
},
{
"depends_on": "eval: [\"Server Script\", \"Webhook\"].includes(doc.task)",
"fetch_if_empty": 1,
"fieldname": "link",
"fieldtype": "Dynamic Link",
"label": "Link",
"mandatory_depends_on": "eval: [\"Server Script\", \"Webhook\"].includes(doc.task)",
"options": "task"
},
{
"default": "0",
"description": "Spawns actions in a background job",
"fieldname": "asynchronous",
"fieldtype": "Check",
"in_list_view": 1,
"in_preview": 1,
"label": "Asynchronous"
}
],
"grid_page_length": 50,
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2025-07-14 12:13:18.108225",
"modified_by": "Administrator",
"module": "Workflow",
"name": "Workflow Transition Task",
"owner": "Administrator",
"permissions": [],
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": []
}

View file

@ -0,0 +1,26 @@
# Copyright (c) 2025, Frappe Technologies and contributors
# For license information, please see license.txt
import frappe
from frappe.model.document import Document
class WorkflowTransitionTask(Document):
# begin: auto-generated types
# This code is auto-generated. Do not modify anything in this block.
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from frappe.types import DF
asynchronous: DF.Check
enabled: DF.Check
link: DF.DynamicLink | None
parent: DF.Data
parentfield: DF.Data
parenttype: DF.Data
task: DF.Literal[None]
# end: auto-generated types
pass

View file

@ -0,0 +1,20 @@
# Copyright (c) 2025, Frappe Technologies and Contributors
# See license.txt
# import frappe
from frappe.tests import IntegrationTestCase
# On IntegrationTestCase, the doctype test records and all
# link-field test record dependencies are recursively loaded
# Use these module variables to add/remove to/from that list
EXTRA_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
IGNORE_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
class IntegrationTestWorkflowTransitionTasks(IntegrationTestCase):
"""
Integration tests for WorkflowTransitionTasks.
Use this class for testing interactions between multiple components.
"""
pass

View file

@ -0,0 +1,19 @@
// Copyright (c) 2025, Frappe Technologies and contributors
// For license information, please see license.txt
frappe.ui.form.on("Workflow Transition Tasks", {
refresh: function (frm) {
frappe
.call({
method: "frappe.workflow.doctype.workflow.workflow.get_workflow_methods",
type: "GET",
})
.then((options) => {
frm.get_field("tasks").grid.update_docfield_property(
"task",
"options",
options.message
);
});
},
});

View file

@ -0,0 +1,46 @@
{
"actions": [],
"allow_rename": 1,
"autoname": "prompt",
"creation": "2025-07-04 15:41:50.296193",
"doctype": "DocType",
"engine": "InnoDB",
"field_order": [
"tasks"
],
"fields": [
{
"fieldname": "tasks",
"fieldtype": "Table",
"label": "Tasks",
"options": "Workflow Transition Task"
}
],
"grid_page_length": 50,
"index_web_pages_for_search": 1,
"links": [],
"modified": "2025-07-08 12:23:01.908648",
"modified_by": "Administrator",
"module": "Workflow",
"name": "Workflow Transition Tasks",
"naming_rule": "Set by user",
"owner": "Administrator",
"permissions": [
{
"create": 1,
"delete": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": []
}

View file

@ -0,0 +1,23 @@
# Copyright (c) 2025, Frappe Technologies and contributors
# For license information, please see license.txt
# import frappe
from frappe.model.document import Document
class WorkflowTransitionTasks(Document):
# begin: auto-generated types
# This code is auto-generated. Do not modify anything in this block.
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from frappe.types import DF
from frappe.workflow.doctype.workflow_transition_task.workflow_transition_task import (
WorkflowTransitionTask,
)
tasks: DF.Table[WorkflowTransitionTask]
# end: auto-generated types
pass