# Extend Runflow

It's impossible Runflow can provide all kinds of tasks you need. When that happens, you can develop your own task and let Runflow load it before a task run.

You will need to write some Python code and hook it up in a Runflow spec.

# Import Flow

If you want to use a flow as part of a new flow, the best way is to import the flow directly.

Say we have a flow examples/template.hcl:

Click me to view the flow `examples/template.hcl`
# File: template.hcl
flow "template" {

  variable "global" {
    default = "global"
  }

  task "hcl2_template" "this" {
    # Set the source for the template.
    # You can interpolate variable via `${...}` syntax.
    source = <<EOT
${ answer }
${ final.answer }
${ var.global }
${ 40 + 2 }
${ sum([20, 20, 2]) }
EOT

    # Set the context for the template rendering.
    # It will be merged with global context.
    context = {
      answer = 42
      final = {
        answer = 42
      }
    }
  }

  task "file_write" "this" {
    filename = "/dev/stdout"
    content = task.hcl2_template.this.content
  }
}

Now, we can import it using import.tasks. The import string for the .hcl file should be a valid Python import string ending with :flow. The key for the import string will be the task type.

For example, the flow below registers examples.template:flow as task type custom_flow_run. The payloads of the task body becomes the variables for the reused flow:

flow "flow_as_task" {

  import {
    tasks = {
      custom_flow_run = "examples.template:flow"
    }
  }

  task "custom_flow_run" "this" {
    global = 42
  }
}
Click me to view the output
$ runflow run examples/flow_as_task.hcl

# Import Task and Function

You can import a custom Task class. The Task class must accept task payload as keyword arguments.

It must has def run(self) or async def run(self) method, which performs the actual task work.

The example below shows how to write something into a file.

# File: examples/extensions.py

class GuessIceCreamTask:

    def __init__(self, name, output):
        self.name = name
        self.output = output

    async def run(self):
        with open(self.output, 'w') as f:
            f.write(f"Bingo, it is {self.name}")

To load it in the Runflow spec, use import.tasks.

# File: custom_task_type.hcl
flow "custom_task_type" {

  import {
    # `import.tasks` is a map.
    # The map key will become the task type used later.
    # The map value is the import string of task implementation.
    tasks = {
      guess_ice_cream = "examples.extensions:GuessIceCreamTask"
    }

    # `import.functions` is a map.
    # The map key will become the function name used later.
    # The map value is the import string of function.
    functions = {
      randint = "random:randint"
    }
  }

  variable "out" {
    default = ""
  }

  task "guess_ice_cream" "echo" {
    name = "${upper("vanilla")}-${randint(1, 100)}"
    output = var.out
  }

}
Click me to view the run output

Run:

$ runflow run custom_task_type.hcl --var out=/tmp/out.txt
[2021-06-13 15:48:35,397] "task.guess_ice_cream.echo" is started.
[2021-06-13 15:48:35,398] "task.guess_ice_cream.echo" is successful.

$ cat /tmp/out.txt
Bingo, it is VANILLA-95

Tips:

  • The Python code for the task must be in sys.path.

# Install as Python Package

The other approach to register a new task class is through the entry_points facility provided by setuptools (opens new window).

Let's demonstrate it through an example.

Create a new directory for your package:

$ mkdir /tmp/runflow_vanilla_example
$ cd /tmp/runflow_vanilla_example
  • Create virtual env.
  • Install build for building your package.
  • Install runflow.
$ python3 -mvenv venv
$ source venv/bin/activate
$ pip install -U build runflow

Create some Python files:

├── src
│   └── runflow_vanilla_example
│       ├── __init__.py
│       └── tasks.py
├── pyproject.toml
└── setup.cfg

Implement your Task class in src/runflow_vanilla_example/tasks.py:

class GuessIceCreamTask:

    def __init__(self, name, output):
        self.name = name
        self.output = output

    async def run(self):
        with open(self.output, 'w') as f:
            f.write(f"Bingo, it is {self.name}")

Export it in src/runflow_vanilla_example/__init__.py:

from .tasks import GuessIceCreamTask

Create a new file setup.cfg as required by setuptools:

  • Define the metadata for the package.
  • Define how to find the package source. In this example, we tell setuptool to find the package source in the directory src/ and include all runflow_vanilla_example source.
[metadata]
name = runflow_vanilla_example
version = 0.1.0
author = Anybody
author_email = anybody@example.org
description = An example package that demonstrates how to register a new task type to Runflow

[options]
package_dir =
    = src
packages = find:

[options.packages.find]
where = src
include =
    runflow_vanilla_example
    runflow_vanilla_example.*

[options.entry_points]
    runflow.tasks =
        guess_ice_cream = runflow_vanilla_example:GuessIceCreamTask

If you look a close look at the last section, it means you register a new task type guess_ice_cream which is implemented by the class GuessIceCreamTask in package runflow_vanilla_example.

Create a new file pyproject.toml:

[build-system]
requires = [
    "setuptools>=42",
    "wheel"
]
build-backend = "setuptools.build_meta"

All source are prepared. Let's build:

$ python -mbuild
...
adding 'runflow_vanilla_example/__init__.py'
adding 'runflow_vanilla_example/tasks.py'
adding 'runflow_vanilla_example-0.1.0.dist-info/METADATA'
adding 'runflow_vanilla_example-0.1.0.dist-info/WHEEL'
adding 'runflow_vanilla_example-0.1.0.dist-info/entry_points.txt'
adding 'runflow_vanilla_example-0.1.0.dist-info/top_level.txt'
adding 'runflow_vanilla_example-0.1.0.dist-info/RECORD'
removing build/bdist.macosx-10.15-x86_64/wheel

$ ls dist/
runflow_vanilla_example-0.1.0.tar.gz
runflow_vanilla_example-0.1.0-py3-none-any.whl

In the directory dist/, two new package files are created.

We can publish the package files to pypi.org (opens new window) using twine (opens new window). The package will be then open to the world.

But for now, let's skip the publish step and just install the package from a local file file to the venv.

$ python -mpip install runflow_vanilla_example-0.1.0-py3-none-any.whl

After runflow_vanilla_example gets installed, Runflow is able to automatically pick it up and recognizes guess_ice_cream as a valid task type.

Hooray!! 🎉

# File: use_package.hcl
flow "use_package" {

  variable "out" {
    default = ""
  }

  task "guess_ice_cream" "echo" {
    name = "${upper("vanilla")}"
    output = var.out
  }

}
Click me to view the run output

Run:

$ runflow run use_package.hcl --var out=/tmp/out.txt
[2021-06-24 23:02:05,109] "task.guess_ice_cream.echo" is started.
[2021-06-24 23:02:05,111] "task.guess_ice_cream.echo" is successful.

$ cat /tmp/out.txt
Bingo, it is VANILLA

# Request to Include Your Implementation

Alternatively, you can issue a new PR to Runflow GitHub repo and request to be included in runflow.registry:task_implementations.

# Which Approach Should I Use?

  1. If you're experimenting something and have .hcl and .py files in one project, just use import block.

  2. If you have pretty solid task implementations and have some tests, docs, consider package all of them and publish to PyPI. This is the most recommended approach.

  3. If you think your implementation is very fundamental and deserved to reside in runflow core library, just send out a PR and let's review if it works!