feat: Workflow Engine

[tanmoysrt]

Sample Code (For screenshots listed below)

README has better explanation.

class DummyBench(WorkflowBuilder):
	# 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

		input_a: DF.Int
		input_b: DF.Int
		output_a: DF.Data | None
		output_b: DF.Data | None
	# end: auto-generated types

	@task
	def add(self) -> int:
		print(f"[add] {self.input_a} + {self.input_b}")
		time.sleep(1)
		result = self.input_a + self.input_b
		print(f"[add] result = {result}")
		return result

	@task
	def multiply(self, a: int, b: int) -> int:
		"""
		Docstring for multiply
		"""

		print(f"[multiply] {a} * {b}")
		time.sleep(1)
		result = a * b
		print(f"[multiply] result = {result}")
		return result

	@task
	def slow_power(self, base: int, exponent: int) -> int:
		"""
		Slow power with base and exponent
		"""
		print(f"[slow_power] {base} ^ {exponent}")

		result = 1
		for i in range(exponent):
			print(f"[slow_power] step {i + 1}/{exponent}...")
			time.sleep(1)
			# Nested call to multiply instead of using '*=' directly
			result = self.multiply.with_task_id(f"power_mult_{i}")(result, base)

		print(f"[slow_power] result = {result}")
		return result

	@task
	def double(self, value: int) -> int:
		print(f"[double] starting doubling {value}")
		time.sleep(1)

		doubled_val = self.multiply.with_task_id("mult")(value, 2)

		print(f"[double] using slow_power to square the doubled value ({doubled_val})")
		final_val = self.slow_power.with_task_id("power")(doubled_val, 2)

		print(f"[double] final stacked result: {final_val}")
		return final_val

	@task
	def double_all(self, values: list[int]) -> list[int]:
		results = []
		print("double all values", values)
		for i, value in enumerate(values):
			print("Trying to double", value)
			doubled = self.double.with_task_id(f"double_{i}")(value)
			print("Doubled", doubled)
			results.append(doubled)
			print(f"[double_all] iteration {i}: double({value}) = {doubled}")
		return results

	@flow
	def run_calculations(self) -> dict:
		sum_result: int = self.add()

		product: int = self.multiply(self.input_a, self.input_b)
		power_result: int = self.slow_power(base=product, exponent=3)

		self.output_a = str(sum_result)
		self.output_b = str(power_result)
		self.save(ignore_permissions=True)

		print(f"[run_calculations] done - output_a={self.output_a}, output_b={self.output_b}")
		return {"output_a": self.output_a, "output_b": self.output_b}

	@flow
	def run_loop_calculations(self, x: int, y: int) -> dict:
		values = [self.input_a, self.input_b, self.input_a + self.input_b]

		results: list[int] = self.double_all(values=values)

		if 5 > 9:
			self.slow_power(12, 12)

		self.add()
		self.double_all.with_task_id("duplicate1")(values=values)

		self.output_a = str(results[0])
		self.output_b = str(results[-1])
		self.save(ignore_permissions=True)

		print(f"[run_loop_calculations] done - results={results}")
		return {"results": results}

Press Workflow

Steps are auto-discovered by AST parsing

Created Tasks

Press Workflow Task

Press Workflow Object

This doctype stores a Python object in a serialized format. For now, using LONGTEXT, ideally should move to BLOB.