repository·main·Indexed 41 days ago
https://github.com/chanzuckerberg/miniwdlA local runner and developer toolkit for the Workflow Description Language (WDL). It enables the execution of WDL workflows locally using container runtimes like Docker or Podman and provides tools for static analysis, linting, and custom function development.
While Docker is the default, miniwdl supports several external plugins for specialized execution environments:
Developers can implement new container backends using the miniwdl Python plugin API. A reference implementation is available in the miniwdl-backend-example repository.
WDL scatter and if (conditional) sections are represented as WorkflowSection objects. These sections create a recursive structure where each section has its own body containing nested nodes.
When analyzing dependencies across these sections, miniwdl uses Gather nodes to represent intermediate dependencies (e.g., when a value inside a scatter is used outside of it, resulting in an Array). To resolve the actual node being depended upon, you can use the final_referee attribute of a Gather node:
# Example of resolving a dependency through a Gather node
dep = workflow.get_node(dep_id)
if isinstance(dep, WDL.Tree.Gather):
dep = dep.final_referee
dep_id = dep.workflow_node_idWhen you execute miniwdl run, the tool creates a run directory to store logs, outputs, and intermediate files.
_LAST is created pointing to the most recent run.--dir /path/to/runs to specify a different root directory.--dir myrun/. to run directly in a specific directory (the directory must be empty or nonexistent).outputs.json in the run directory is a reliable indicator of a successful run. The JSON output is also printed to stdout.error.json is written to the run directory. If the --error-json flag is used, this error information is also printed to stdout.outputs.json nor error.json exists, the run is likely still in progress (unless the process was force-killed).For Tasks:
task.log: Task execution log.write_/: Files created via WDL expressions (e.g., write_lines()).download/: Files downloaded from URIs in task inputs.work/: The working directory mounted into the container.stdout.txt / stderr.txt: Streamed command output.out/: Symbolic links to individual output files organized by the WDL output declarations.For Workflows:
workflow.log: Workflow execution log.write_/ and download/: Similar to tasks.out/: Links reaching into call subdirectories where outputs were generated.Top-level Run Directory:
wdl/: A copy of the original WDL and its imports.rerun: A file that can be "sourced" to re-run the WDL using the same inputs.# Run in a specific directory without creating a timestamped subdirectory
miniwdl run --dir myrun/. my_workflow.wdl
# Run and print error JSON to stdout on failure
miniwdl run --error-json my_workflow.wdlPolymorphic functions (where the return type depends on the input types, such as choose_random(Array[T]) -> T) cannot use the @static() helper. Instead, you must implement a class derived from WDL.StdLib.EagerFunction and assign an instance of it to the function name in WDL.StdLib.Base.__init__.
An eager function requires two main methods:
infer_type(self, expr: "Expr.Apply") -> Type.Base: Invoked during static WDL validation. It inspects the argument expressions to determine the return type or raises Error.WrongArity or Error.StaticTypeMismatch._call_eager(self, expr: "Expr.Apply", arguments: List[Value.Base]) -> Value.Base: Invoked at runtime once arguments are evaluated. It performs the actual logic and returns the result wrapped in a WDL.Value (e.g., Value.Pair).class _ChooseRandom(EagerFunction):
def infer_type(self, expr: "Expr.Apply") -> Type.Base:
# Logic to check arity and return Type.Array.item_type or Type.Pair
...
def _call_eager(self, expr: "Expr.Apply", arguments: List[Value.Base]) -> Value.Base:
# Logic to pick random items and return Value.Pair or Value.Base
...The download cache stores URI-based inputs (like reference genomes) in a central directory so they can be reused across different tasks and workflows without re-downloading. This allows you to use portable URIs in your WDL instead of local paths.
Key behaviors:
atime) and opening a shared flock() on files currently in use.miniwdl localize subcommand to "prime" the cache with URIs from a JSON input template before running a workflow.--no-cache flag to disable caching for a single run.To enable it, add a [download_cache] section to your configuration.
[download_cache]
put = true
get = true
dir = /tmp/miniwdl_download_cacheWhen implementing WDL functions that require file access (e.g., word_count(text_file)), use the self._devirtualize_filename() method. This method is an abstract method of the WDL.StdLib.Base object that converts a WDL File value into a local filename that the Python process can open. This handles complexities like translating paths from a Docker container to the host file system.
@static([Type.File()], Type.Int())
def word_count(v: Value.File) -> Value.Int:
with open(self._devirtualize_filename(v.value), "r") as infile:
return Value.Int(len(infile.read().split(" ")))The WorkflowNode abstraction is the primary way to traverse a WDL workflow's dependency structure. The Workflow.body attribute contains a list of objects deriving from WorkflowNode (such as Call, Decl, Scatter, and Conditional sections).
Key attributes of a WorkflowNode include:
workflow_node_id: A unique string identifier for the node.workflow_node_dependencies: A set of node IDs that the current node depends on (pre-computed via static analysis).name: The name of the call or declaration.This abstraction is more convenient for dependency analysis than traversing the raw syntax tree.
Static functions have fixed argument and return types. They are implemented in WDL/StdLib.py using the @static() helper decorator within the WDL.StdLib.Base class.
Key details:
@static() decorator takes a list of argument WDL types (WDL.Type.Base) and a return type.WDL.Value objects (e.g., Value.Int).WDL.Value object.pyre for type checking.@static([Type.Int()], Type.Int())
def factorial(v: Value.Int) -> Value.Int:
def f(n: int) -> int:
return 1 if n <= 1 else n * f(n - 1)
return Value.Int(f(v.value))The call cache allows miniwdl to reuse task and workflow outputs for repeat calls with the same inputs. This is useful for debugging or resuming from errors.
Key behaviors:
*.json files in the configured directory.--no-cache flag to disable caching for a single run.To enable it, add a [call_cache] section to your configuration.
[call_cache]
put = true
get = true
dir = ~/.cache/miniwdlBy default, miniwdl run creates a timestamp-prefixed subdirectory in the current working directory for all operations.
dir). This is also saved as outputs.json inside the run subdirectory._LAST): miniwdl creates a symbolic link named _LAST pointing to the most recent run's subdirectory.out/): Inside the run subdirectory, an out directory contains symbolic links to the actual output files. This is often the most convenient way to consume file outputs.If your run directory is 20200604_132146_assemble_refbased, you can find files in _LAST/out/:
_LAST/out/
├── assembly_fasta
│ └── G5012.3.fasta -> ../../call-call_consensus/work/G5012.3.fasta
└── ...miniwdl can be scaled to handle larger workloads using the following approaches:
The omnibus example serves as a reference for implementing the following WDL 1.2 capabilities:
File and Directory declarations.main.wdl to modules in lib/).Directory inputs and outputs.join_paths.requirements and hints sections instead of the deprecated runtime block.task.name, task.container, and task.return_code.env inputs, including serializing struct types into environment variables.struct definitions using meta and parameter_meta.Map and Optional types.find, matches, contains, prefix, suffix, quote, squote.chunk, join_paths, values, length(String|Map), as_map, as_pairs, collect_by_key, zip, unzip, cross.select_first, contains_key.read_tsv (including header overloads).input: keyword.** operator.