From 6504387c8bc05d8321694fc0468b605a00f3078b Mon Sep 17 00:00:00 2001 From: Apeksha_2022 <70309060+Apeksha9389@users.noreply.github.com> Date: Sat, 10 Oct 2026 09:35:35 +0530 Subject: [PATCH 1/2] feat: allow building ConductorWorkflow without an executor --- .../client/workflow/conductor_workflow.py | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/conductor/client/workflow/conductor_workflow.py b/src/conductor/client/workflow/conductor_workflow.py index 7ab521ec6..6df2ea3a3 100644 --- a/src/conductor/client/workflow/conductor_workflow.py +++ b/src/conductor/client/workflow/conductor_workflow.py @@ -25,10 +25,12 @@ class ConductorWorkflow: SCHEMA_VERSION = 2 def __init__(self, - executor: WorkflowExecutor, - name: str, + executor: Optional[WorkflowExecutor] = None, + name: Optional[str] = None, version: Optional[int] = None, description: Optional[str] = None) -> Self: + if name is None: + raise TypeError("ConductorWorkflow() missing required argument: 'name'") self._executor = executor self.name = name self.version = version @@ -206,8 +208,16 @@ def workflow_input(self, input: dict) -> Self: # Register the workflow definition with the server. If overwrite is set, the definition on the server will be # overwritten. When not set, the call fails if there is any change in the workflow definition between the server # and what is being registered. + def _require_executor(self) -> WorkflowExecutor: + if self._executor is None: + raise RuntimeError( + "This ConductorWorkflow has no executor. Pass executor= when creating it " + "to register, start or execute the workflow." + ) + return self._executor + def register(self, overwrite: bool): - return self._executor.register_workflow( + return self._require_executor().register_workflow( overwrite=overwrite, workflow=self.to_workflow_def(), ) @@ -226,7 +236,7 @@ def start_workflow(self, start_workflow_request: StartWorkflowRequest) -> str: start_workflow_request.workflow_def = self.to_workflow_def() start_workflow_request.name = self.name start_workflow_request.version = self.version - return self._executor.start_workflow(start_workflow_request) + return self._require_executor().start_workflow(start_workflow_request) def start_workflow_with_input(self, workflow_input: Optional[dict] = None, correlation_id: Optional[str] = None, task_to_domain: Optional[Dict[str, str]] = None, priority: Optional[int] = None, idempotency_key: Optional[str] = None, idempotency_strategy: IdempotencyStrategy = IdempotencyStrategy.FAIL) -> str: @@ -245,7 +255,7 @@ def start_workflow_with_input(self, workflow_input: Optional[dict] = None, corre start_workflow_request.priority = priority start_workflow_request.task_to_domain =task_to_domain - return self._executor.start_workflow(start_workflow_request) + return self._require_executor().start_workflow(start_workflow_request) def execute(self, workflow_input: Any = None, wait_until_task_ref: str = "", wait_for_seconds: int = 10, request_id: Optional[str] = None, @@ -274,8 +284,8 @@ def execute(self, workflow_input: Any = None, wait_until_task_ref: str = "", wai request.idempotency_strategy = idempotency_strategy if task_to_domain is not None: request.task_to_domain = task_to_domain - run = self._executor.execute_workflow(request, wait_until_task_ref=wait_until_task_ref, - wait_for_seconds=wait_for_seconds, request_id=request_id) + run = self._require_executor().execute_workflow(request, wait_until_task_ref=wait_until_task_ref, + wait_for_seconds=wait_for_seconds, request_id=request_id) return run From f074b678fbdc2495c56799d4f165aaa3516c325c Mon Sep 17 00:00:00 2001 From: Apeksha_2022 <70309060+Apeksha9389@users.noreply.github.com> Date: Sat, 10 Oct 2026 09:35:52 +0530 Subject: [PATCH 2/2] test: cover ConductorWorkflow without an executor --- .../test_conductor_workflow_offline.py | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 tests/unit/workflow/test_conductor_workflow_offline.py diff --git a/tests/unit/workflow/test_conductor_workflow_offline.py b/tests/unit/workflow/test_conductor_workflow_offline.py new file mode 100644 index 000000000..56cf6d90e --- /dev/null +++ b/tests/unit/workflow/test_conductor_workflow_offline.py @@ -0,0 +1,45 @@ +import unittest +from unittest.mock import MagicMock + +from conductor.client.workflow.conductor_workflow import ConductorWorkflow +from conductor.client.workflow.task.simple_task import SimpleTask + + +class TestConductorWorkflowWithoutExecutor(unittest.TestCase): + def test_build_definition_without_executor(self): + workflow = ConductorWorkflow(name="offline_wf", version=2) + workflow.add(SimpleTask(task_def_name="task_a", task_reference_name="a")) + + workflow_def = workflow.to_workflow_def() + self.assertEqual(workflow_def.name, "offline_wf") + self.assertEqual(workflow_def.version, 2) + self.assertEqual([t.task_reference_name for t in workflow_def.tasks], ["a"]) + + def test_name_is_still_required(self): + with self.assertRaises(TypeError): + ConductorWorkflow() + + def test_register_without_executor_raises(self): + with self.assertRaisesRegex(RuntimeError, "executor="): + ConductorWorkflow(name="offline_wf").register(overwrite=True) + + def test_start_workflow_with_input_without_executor_raises(self): + with self.assertRaises(RuntimeError): + ConductorWorkflow(name="offline_wf").start_workflow_with_input({}) + + def test_execute_without_executor_raises(self): + with self.assertRaises(RuntimeError): + ConductorWorkflow(name="offline_wf").execute({}) + + def test_positional_executor_still_used(self): + executor = MagicMock() + workflow = ConductorWorkflow(executor, "wf", 1) + + workflow.register(overwrite=True) + + executor.register_workflow.assert_called_once() + self.assertEqual(executor.register_workflow.call_args.kwargs["workflow"].name, "wf") + + +if __name__ == "__main__": + unittest.main()