Stages

Stages module include all stage model that use be the minimum execution layer of this workflow engine. The stage handle the minimize task that run in some thread (same thread at its job owner) that mean it is the lowest executor that you can track logs.

The output of stage execution only return SUCCESS or CANCEL status because I do not want to handle stage error on this stage execution. I think stage model have a lot of use-case, and it should does not worry about it error output.

So, I will create handler_execute for any exception class that raise from the stage execution method.

Execution   ┬-> Ok      ---( handler )--> Result with SUCCESS/CANCEL
            │
            ╰-> Error   ┬--( handler )--> Result with FAILED (Set `raise_error` flag)
                        │
                        ╰--( handler )--> Raise StageException(...)

On the context I/O that pass to a stage object at execute process. The execute method receives a params={"params": {...}} value for mapping to template searching.

All stages model inherit from BaseStage or AsyncBaseStage models that has the base fields:

field	alias	data type	default	description
id		str | None	None	A stage ID that use to keep execution output or getting by job owner.
name		str		A stage name that want to logging when start execution.
condition	if	str | None	None	A stage condition statement to allow stage executable.
extras		dict	dict()	An extra parameter that override core config values.

Empty Stage

Empty stage executor that do nothing and log the message field to stdout only. It can use for tracking a template parameter on the workflow or debug step.

You can pass a sleep value in second unit to this stage for waiting after log message.

YAML

EchoEcho with IDSleep

stages:
  - name: Echo hello world
    echo: "Hello World"

stages:
  - name: Echo hello world
    id: echo-stage
    echo: "Hello World"

stages:
  - name: Echo hello world
    id: echo-sleep-stage
    echo: "Hello World and Sleep 10 seconds"
    sleep: 10

field	data type	default	description
echo	str | None	None	A message that want to show on the stdout.
sleep	float	0	A second value to sleep before start execution. This value should gather or equal 0, and less than 1800 seconds.

Bash Stage

Bash execution stage that execute bash script on the current OS. If your current OS is Windows, it will run on the bash in the WSL.

Warning

I get some limitation when I run shell statement with the built-in subprocess package. It does not good enough to use multiline statement. Thus, I add writing .sh file before execution process for fix this issue.

YAML

BashBash with Env

stages:
    - name: Call echo
      bash: |
        echo "Hello Bash Stage";

stages:
    - name: Call echo with env
      env:
        FOO: BAR
      bash: |
        echo "Hello Bash $FOO";

field	data type	default	description
bash	str		A bash statement that want to execute via Python subprocess.
env	dict[str, Any]	dict()	An environment variables that set before run bash command. It will add on the header of the .sh file.

Python Stage

Python stage that running the Python statement with the current globals and passing an input additional variables via exec built-in function.

This stage allow you to use any Python object that exists on the globals such as import your installed package.

Warning

The exec build-in function is very dangerous. So, it should use the re module to validate exec-string before running or exclude the os package from the current globals variable.

YAML

Python

stages:
    - name: Call Python
      run: |
        import sys
        print(sys.version_info)

field	data type	default	description
run	str		A Python string statement that want to run with exec.
vars	dict[str, Any]	dict()	A variable mapping that want to pass to globals parameter in the exec func.

Call Stage

Call stage executor that call the Python function from registry with tag decorator function in reusables module and run it with input arguments.

This stage is different with PyStage because the PyStage is just run a Python statement with the exec function and pass the current locals and globals before exec that statement. This stage will import the caller function can call it with an input arguments. So, you can create your function complexly that you can for your objective to invoked by this stage object.

This stage is the most powerfull stage of this package for run every use-case by a custom requirement that you want by creating the Python function and adding it to the caller registry value by importer syntax like module.caller.registry not path style like module/caller/registry.

Warning

The caller registry to get a caller function should importable by the current Python execution pointer.

YAML

Call

stages:
    - name: Call call task
      uses: tasks/el-csv-to-parquet@polars
      with:
        source-path: "./data"
        target-path: "./warehouse"

field	alias	data type	default	description
uses		str		A caller function with registry importer syntax that use to load function before execute step. The caller registry syntax should be <import.part>/<func-name>@<tag-name>.
args	with	dict[str, Any]	dict()	An argument parameter that will pass to this caller function.

Trigger Stage

Trigger workflow executor stage that run an input trigger Workflow execute method. This is the stage that allow you to create the reusable Workflow template with dynamic parameters.

YAML

Trigger

stages:
    - name: Call trigger
      trigger: trigger-workflow-name
      params:
        name: foo

field	data type	default	description
trigger	str		A trigger workflow name that should already exist on the config.
params	dict[str, Any]	dict()	A parameter that want to pass to workflow execution.

Parallel Stage

Parallel stage executor that execute branch stages with multithreading. This stage let you set the fix branches for running child stage inside it on multithread pool.

Note

This stage is not the low-level stage model because it runs multi-stages in this stage execution.

field	alias	data type	default	description
parallel		dict[str, list[Stage]]		A mapping of branch name and its stages.
max_workers	max-workers	int	2	The maximum multi-thread pool worker size for execution parallel. This value should be gather or equal than 1, and less than 20.

ForEach Stage

For-Each stage executor that execute all stages with each item in the foreach list.

Note

This stage is not the low-level stage model because it runs multi-stages in this stage execution.

Until Stage

Until stage executor that will run stages in each loop until it valid with stop loop condition.

Note

This stage is not the low-level stage model because it runs multi-stages in this stage execution.

Case Stage

Case stage executor that execute all stages if the condition was matched.

Raise Stage

Raise error stage executor that raise StageException that use a message field for making error message before raise.

Docker Stage

Docker container stage execution that will pull the specific Docker image with custom authentication and run this image by passing environment variables and mounting local volume to this Docker container.

The volume path that mount to this Docker container will limit. That is this stage does not allow you to mount any path to this container.

Virtual Python Stage

Virtual Python stage executor that run Python statement on the dependent Python virtual environment via the uv package.