Add workflow_task_list parameter to WorkflowTask

docs/src/features/task_lists.md

@@ -65,3 +65,42 @@ class MyWorkflow(Workflow):
 
 In this example, `task_list` is a mandatory workflow argument; a more realistic
 case would use a `kwarg`.
+
+## Child workflows
+
+By default, a child workflow is launched on the task list determined by its own
+`task_list` class variable or `get_task_list` class method, just like a
+top-level workflow. To override this on a per-submission basis — for instance to
+make a child workflow inherit its parent's decision task list — pass
+`workflow_task_list` when building the `WorkflowTask`:
+
+```python
+from simpleflow import Workflow
+from simpleflow.task import WorkflowTask
+
+
+class MyChildWorkflow(Workflow):
+    ...
+
+
+class MyWorkflow(Workflow):
+    ...
+
+    def run(self, *args, **kwargs):
+        # Run the child workflow on this workflow's own decision task list.
+        self.submit(
+            WorkflowTask(
+                self._executor,
+                MyChildWorkflow,
+                *args,
+                workflow_task_list=self._executor.task_list,
+            )
+        )
+```
+
+`workflow_task_list` takes precedence over the child workflow's own
+`task_list`/`get_task_list`. It is purely a decision-routing concern: it is
+never forwarded as an argument to the child workflow's `run()`.
+
+!!! warning
+    `workflow_task_list` is a reserved keyword argument of `WorkflowTask`.

simpleflow/swf/task.py

@@ -161,7 +161,14 @@ def from_generic_task(cls, task: task.WorkflowTask) -> Self:
         """
         Casts a generic simpleflow.task.WorkflowTask into a SWF one.
         """
-        return cls(task.executor, task.workflow, *task._args, **task._kwargs)
+        # Preserve an explicit decision task list override set on the generic task.
+        return cls(
+            task.executor,
+            task.workflow,
+            *task._args,
+            workflow_task_list=task._task_list,
+            **task._kwargs,
+        )
 
     @property
     def name(self):
@@ -173,6 +180,11 @@ def payload(self):
 
     @property
     def task_list(self) -> str | None:
+        # An explicit override (e.g. to make a child workflow inherit its
+        # parent's decision task list) wins over the workflow's own logic.
+        if self._task_list:
+            return self._task_list
+
         get_task_list = getattr(self.workflow, "get_task_list", None)
         if get_task_list:
             return get_task_list(self.workflow, *self.args, **self.kwargs)

simpleflow/task.py

@@ -146,7 +146,14 @@ class WorkflowTask(Task):
     Child workflow.
     """
 
-    def __init__(self, executor: Executor | None, workflow: type[Workflow], *args, **kwargs) -> None:
+    def __init__(
+        self,
+        executor: Executor | None,
+        workflow: type[Workflow],
+        *args,
+        workflow_task_list: str | None = None,
+        **kwargs,
+    ) -> None:
         # Keep original arguments for use in subclasses
         # For instance this helps casting a generic class to a simpleflow.swf.task,
         # see simpleflow.swf.task.WorkflowTask.from_generic_task() factory
@@ -159,6 +166,11 @@ def __init__(self, executor: Executor | None, workflow: type[Workflow], *args, *
         get_workflow_id = getattr(workflow, "get_workflow_id", None)
         self.args = self.resolve_args(*args)
         self.kwargs = self.resolve_kwargs(**kwargs)
+        # Explicit decision task list for the child workflow. When set, it takes
+        # precedence over the child workflow's own ``get_task_list``/``task_list``
+        # (see simpleflow.swf.task.WorkflowTask.task_list). It is *not* a run
+        # argument: it never reaches the child's ``run()``.
+        self._task_list = workflow_task_list
 
         if get_workflow_id:
             self.id: str | None = get_workflow_id(workflow, *self.args, **self.kwargs)
@@ -192,8 +204,8 @@ class ChildWorkflowTask(WorkflowTask):
     (yet).
     """
 
-    def __init__(self, workflow: type[Workflow], *args, **kwargs) -> None:
-        super().__init__(None, workflow, *args, **kwargs)
+    def __init__(self, workflow: type[Workflow], *args, workflow_task_list: str | None = None, **kwargs) -> None:
+        super().__init__(None, workflow, *args, workflow_task_list=workflow_task_list, **kwargs)
 
 
 class SignalTask(Task):

tests/test_simpleflow/swf/test_task.py

@@ -2,8 +2,8 @@
 
 from sure import expect
 
-from simpleflow import activity
-from simpleflow.swf.task import ActivityTask
+from simpleflow import Workflow, activity, task
+from simpleflow.swf.task import ActivityTask, WorkflowTask
 
 
 @activity.with_attributes()
@@ -32,3 +32,64 @@ def test_task_attaches_context_to_object_instances():
     ctx = {"foo": "bar"}
     expect(ActivityTask(ShowContextCls, context=ctx).execute()).to.equal(ctx)
     expect(ShowContextCls.context).to.be.none
+
+
+class PlainWorkflow(Workflow):
+    name = "plain_workflow"
+
+    def run(self, *args, **kwargs):
+        pass
+
+
+class TaskListWorkflow(Workflow):
+    name = "task_list_workflow"
+    task_list = "from_attribute"
+
+    def run(self, *args, **kwargs):
+        pass
+
+
+class GetTaskListWorkflow(Workflow):
+    name = "get_task_list_workflow"
+
+    @staticmethod
+    def get_task_list(workflow, *args, **kwargs):
+        return "from_get_task_list"
+
+    def run(self, *args, **kwargs):
+        pass
+
+
+def test_task_list_defaults_to_none():
+    expect(WorkflowTask(None, PlainWorkflow).task_list).to.be.none
+
+
+def test_task_list_falls_back_to_workflow_attribute():
+    expect(WorkflowTask(None, TaskListWorkflow).task_list).to.equal("from_attribute")
+
+
+def test_task_list_falls_back_to_get_task_list():
+    expect(WorkflowTask(None, GetTaskListWorkflow).task_list).to.equal("from_get_task_list")
+
+
+def test_explicit_task_list_wins_over_get_task_list():
+    wf_task = WorkflowTask(None, GetTaskListWorkflow, workflow_task_list="explicit")
+    expect(wf_task.task_list).to.equal("explicit")
+
+
+def test_explicit_task_list_wins_over_workflow_attribute():
+    wf_task = WorkflowTask(None, TaskListWorkflow, workflow_task_list="explicit")
+    expect(wf_task.task_list).to.equal("explicit")
+
+
+def test_from_generic_task_preserves_explicit_task_list():
+    generic = task.WorkflowTask(None, GetTaskListWorkflow, workflow_task_list="explicit")
+    swf_task = WorkflowTask.from_generic_task(generic)
+    expect(swf_task).to.be.a(WorkflowTask)
+    expect(swf_task.task_list).to.equal("explicit")
+
+
+def test_from_generic_task_without_explicit_task_list():
+    generic = task.WorkflowTask(None, GetTaskListWorkflow)
+    swf_task = WorkflowTask.from_generic_task(generic)
+    expect(swf_task.task_list).to.equal("from_get_task_list")

tests/test_simpleflow/test_task.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from simpleflow import activity, registry, task
+from simpleflow import Workflow, activity, registry, task
 
 
 @activity.with_attributes(task_list="test")
@@ -33,3 +33,36 @@ def test_task_register():
     _registry = registry.registry[None]
     assert _registry["tests.test_simpleflow.test_task.double"] == double
     assert _registry["tests.test_simpleflow.test_task.Double"] == Double
+
+
+class MyWorkflow(Workflow):
+    name = "my_workflow"
+
+    def run(self, *args, **kwargs):
+        return args, kwargs
+
+
+def test_workflow_task_stores_explicit_task_list():
+    wf_task = task.WorkflowTask(None, MyWorkflow, 1, foo="bar", workflow_task_list="my_list")
+    assert wf_task._task_list == "my_list"
+
+
+def test_workflow_task_list_defaults_to_none():
+    wf_task = task.WorkflowTask(None, MyWorkflow, 1, foo="bar")
+    assert wf_task._task_list is None
+
+
+def test_workflow_task_list_does_not_leak_into_run_arguments():
+    # The explicit task list is a decision-routing concern; it must never be
+    # forwarded as an argument to the child workflow's run().
+    wf_task = task.WorkflowTask(None, MyWorkflow, 1, foo="bar", workflow_task_list="my_list")
+    assert wf_task.args == [1]
+    assert wf_task.kwargs == {"foo": "bar"}
+    assert "workflow_task_list" not in wf_task._kwargs
+
+
+def test_child_workflow_task_forwards_explicit_task_list():
+    wf_task = task.ChildWorkflowTask(MyWorkflow, 1, foo="bar", workflow_task_list="my_list")
+    assert wf_task._task_list == "my_list"
+    assert wf_task.args == [1]
+    assert wf_task.kwargs == {"foo": "bar"}