Workflow

Dynamically build workflow definitions based on the inputs and hints

Declaration

You might want to dynamically generate your workflow specification based on the inputs and hints.

Note

Dynamic workflows produce a static workflow definition and cannot be used to conditionally run steps based on the outputs of another step. For that, please see the *conditionals* documentation.

Other important comments:

  • DynamicWorkflows can only be run directly - they cannot be imported from a registry.
  • You must override the constructor argument.
  • You could achieve the similar functionality with a WorkflowBuilder, but the DynamicWorkflow class allows JanisAssistant to process the inputs and submit to an engine for you.
class janis.DynamicWorkflow(**connections)[source]
constructor(inputs: Dict[str, any], hints: Dict[str, str])[source]

A place to construct your workflows. This is called after inputs are initially processed

Parameters:
  • inputs – Dictionary of input values
  • hints – Dictionary of hints that should be applied
modify_inputs(inputs, hints: Dict[str, str])[source]
Parameters:
  • inputs – Dictionary of input values
  • hints – Dictionary of hints that should be applied
Returns:

The modified inputs dictionary

Example

We’ll create a workflow that takes an input called inp that takes a string or an array of strings and we’ll create a workflow based on the input.

class MyFirstDynamicWorkflow(DynamicWorkflow):
    def friendly_name(self):
        return "My first dynamic workflow"

    def id(self) -> str:
        return "MyFirstDynamicWorkflow"

    def constructor(self, inputs: Dict[str, any], hints: Dict[str, str]):

        my_value_to_print = inputs.get("inp")

        if isinstance(my_value_to_print, list):
            # print each element if 'inp' is a list
            self.input("inp_list", Array(String))
            self.step("print", Echo(inp=self.inp_list), scatter="inp")
        else:
            # print only once
            self.input("inp", String)
            self.step("print", Echo(inp=self.inp))

        # Janis will derive the output from the source (Array(String) | String)
        self.output("out", source=self.print)

    def modify_inputs(self, inputs, hints: Dict[str, str]):
        my_value_to_print = inputs.get("inp")
        if isinstance(my_value_to_print, list):
            return {**inputs, "inp_list": my_value_to_print}
        else:
            return inputs

We’ll run the following command to generate WDL of the above workflow:

janis translate --inputs input.yaml dynamicworkflow.py wdl

Using an input.yaml with single string, we get the following WDL:

workflow MyFirstDynamicWorkflow {
  input {
    String inp
  }
  call E.echo as print {
    input:
      inp=inp
  }
  output {
    File out = print.out
  }
}

Using an input.yaml with an array of strings, we get the following WDL:

workflow MyFirstDynamicWorkflow {
  input {
    Array[String] inp_list
  }
  scatter (i in inp_list) {
     call E.echo as print {
      input:
        inp=i
    }
  }
  output {
    Array[File] out = print.out
  }
}