Add ability to resolve workflows within applications, closes #290

src/python/txtai/app/base.py

@@ -242,9 +242,14 @@ class Application:
             resolved function
         """
 
+        # Check if function is a pipeline
         if function in self.pipelines:
             return self.pipelines[function]
 
+        # Check if function is a workflow
+        if function in self.workflows:
+            return self.workflows[function]
+
         # Attempt to resolve action as a callable function
         return PipelineFactory.create({}, function)
 

test/python/testworkflow.py

@@ -477,6 +477,30 @@ class TestWorkflow(unittest.TestCase):
             "And another sentence to split.",
         )
 
+    def testYamlWorkflowTask(self):
+        """
+        Tests YAML workflow with a workflow task
+        """
+
+        # Create function and add to module
+        def action(elements):
+            return [x * 2 for x in elements]
+
+        sys.modules[__name__].action = action
+
+        workflow = """
+        workflow:
+            run:
+                tasks:
+                    - testworkflow.action
+            flow:
+                tasks:
+                    - run
+        """
+
+        app = API(workflow)
+        self.assertEqual(list(app.workflow("flow", [1, 2])), [2, 4])
+
     def testYamlTransformWorkflow(self):
         """
         Test reading a YAML transform workflow in Python.