feat: allow building ConductorWorkflow without an executor

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
 

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()