How to move the context into another function

[psmit]

Hi,

I have a graph based worklow structure (already created in python), and I want to transform this into an Argo DAG. In principle, all the examples are clear, but none of them use external functions (or recursive functions). As the context manager isn't used directly, there is no context to pass on.

For example, I have the following graph, with only a single leaf:

class MyGraph(object):
    def __init__(inputs: list[MyGraph]):
        self.inputs = inputs

and I would like to write code like:

def submit(graph):
   with DAG():
       submit_inner(graph)
       
def submit_inner(graph):
    G = Container()
    deps = [submit_inner(i) for i in self.inputs]
    if len(deps) > 0:
        deps >> G
    return G

Would it be possible to write recursive code like this with the current API?

[elliotgunton]

Hey @psmit - the context manager is implicitly alive during any function calls under a context. However, in your example you would need to be in a Workflow context to be able to define a DAG otherwise the Tasks will try to add the templates they use to a Workflow context that doesn't exist. It's also not currently possible for a list of Tasks to be the first item in a chain of dependencies - you would have to use a workaround to assign next for each task in the list. For a Workflow example using a function to create Tasks and this workaround, see below, hope this helps!

def create_tasks(names: List[str], container: Container) -> List[Task]:
    return [Task(name=f"container-in-function-{n}", template=container) for n in names]


def submit_workflow():
    with Workflow(
        generate_name="context-example-",
        entrypoint="dag-template",
    ) as w:
        container = Container(name="container", command=["echo", "test"], image="docker/whalesay:latest")
        with DAG(name="dag-template"):
            tasks = create_tasks(["task-1", "task-2"], container)
            task_in_context = container(name="container-in-context")
            for t in tasks:
                t.next(task_in_context)

    w.create()

You can also check the yaml generated from this with print(w.to_yaml()):

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: context-example-
spec:
  entrypoint: dag-template
  templates:
  - container:
      command:
      - echo
      - test
      image: docker/whalesay:latest
    name: container
  - dag:
      tasks:
      - name: container-in-function-task-1
        template: container
      - name: container-in-function-task-2
        template: container
      - depends: container-in-function-task-1 && container-in-function-task-2
        name: container-in-context
        template: container
    name: dag-template