Winslow

A workflow and state manager with a terminal UI.

Winslow runs work as a set of small tasks. Each task declares the tasks it depends on. Winslow builds the dependency graph and runs the tasks in the correct order.

Each task knows two things: how to do its work, and how to report that the work is already done. Winslow checks the second before it does the first.

The core idea: run and check

Two methods form the whole contract:

• run() makes a change.
• check() reports whether the wanted end state is already true.

Winslow calls check() before it runs a task. If check() returns true, Winslow marks the task completed and does not run it. If it returns false, Winslow calls run(), then calls check() again to confirm the result.

This makes a workflow idempotent and resumable. Stop a workflow in the middle and start it again. Winslow continues from the point it reached.

Adopt Winslow one task at a time

The run() method is optional. The check() method is not. A task that omits run() changes nothing and only reports a state.

A workflow built only from such tasks is a health check tool. It reads a system and it writes nothing. Two uses follow from this:

A status view over a legacy system. Describe the wanted state of the system as a set of checks. The terminal UI then shows which parts of the system are correct, and which parts are not. The legacy process continues to run as before. Winslow only observes it.

From a report to an automation. Add a run() method to one task at a time. Winslow automates that task from the moment that the method exists. The other tasks stay as checks. The workflow is usable at every step of the migration, and no step needs a large change.

Install

uv add "winslow[tui]"      # or: pip install 'winslow[tui]'

Winslow needs Python 3.12 or later. The tui extra installs the terminal UI. For headless runs in cron or CI, the winslow package alone is sufficient.

Quick start

A workflow is a directory. It holds a Workflow class and the Task classes that belong to that workflow. The filename workflow.py marks the directory as a workflow package. Every .py file beside it belongs to the same workflow.

Put this content in workflows/etl/workflow.py:

examples/etl/workflow.py

from pathlib import Path

from winslow import Workflow, Task

RAW = Path("raw.txt")
CLEAN = Path("clean.txt")


class Etl(Workflow):
    pass  # The name defaults to "etl", the kebab-cased class name.


class DownloadData(Task):
    def run(self):
        RAW.write_text("oslo 4\nlisbon 17\n")

    def check(self):
        return RAW.exists()


class TransformData(Task):
    dependencies = DownloadData

    def run(self):
        warm = [ln for ln in RAW.read_text().splitlines() if int(ln.split()[1]) > 10]
        CLEAN.write_text("\n".join(warm))

    def check(self):
        return CLEAN.exists()

Download this example

Project layout

The example holds the workflow and the tasks in one file, which keeps it short. A real workflow puts the tasks in their own files beside workflow.py. Winslow imports every .py file in the directory. See Workflows.

Start the terminal UI from that directory:

winslow run

Or run the workflow headless:

winslow run --mode headless --workflow etl

Run the same command a second time. Winslow runs neither task, because check() already returns true.

Pass options to a workflow

A workflow declares its runtime options with ConfigOption. Each option becomes a command line argument. A task reads the values from self.workflow_config.

examples/report/workflow.py

from pathlib import Path

from winslow import Workflow, Task, ConfigOption


class Report(Workflow):
    region = ConfigOption(
        type=str,
        required=True,
        choices=["eu", "us"],
        help_text="The region to report on.",
    )
    limit = ConfigOption(type=int, default=10, help_text="The maximum row count.")


class WriteReport(Task):
    @property
    def output_path(self):
        return Path(f"report-{self.workflow_config.region}.txt")

    def run(self):
        self.output_path.write_text(f"rows: {self.workflow_config.limit}\n")

    def check(self):
        return self.output_path.exists()

Download this example

Pass each value on the command line:

winslow run --mode headless --workflow report --region eu --limit 5

An option name uses an underscore in Python and a dash on the command line. The option max_rows thus becomes --max-rows.

The region option is required. Winslow stops before it runs a task if the command omits the value:

Workflow - report: error: the following arguments are required: --region

The limit option has a default, so the command can omit it. The choices list is also enforced:

Workflow - report: error: argument --region: invalid choice: 'xx' (choose from eu, us)

The terminal UI presents the same options as a form. Fill the form in, then start the workflow.

The environment

Winslow reads the environment name from the WINSLOW_ENV variable, and the default value is dev. A task and a workflow both read the value from self.env.

Filters

Run or check a subset of the tasks with a filter expression:

winslow run --filter test    # Every task whose name contains "test".

A filter also selects by group, and the operators combine the selections. The search box of the terminal UI accepts the same language. See Filters.

Trust model

Winslow runs the code in the directory that you start it from

Treat a workflow directory like a Makefile or a conftest.py. Start Winslow only in a directory that you trust.

At startup Winslow searches the current directory and every subdirectory below it for a workflow.py file. It then imports each workflow.py file, and every other .py file in the same directory tree. Winslow ignores a directory whose name starts with a full stop or an underscore. These imports happen before the first prompt.

Where to go next

• Workflows: declare a workflow, arrange the task files, and share code between workflows.
• Tasks: the task lifecycle, and the method that each stage calls.
• Dependencies: order the tasks with the dependency graph, premier and terminal tasks, and groups.
• Filters: select a subset of the tasks by name or by group.
• Parameterization: turn one task class into many task instances.
• Constraints: package a gate rule as a class, and share it between tasks.
• Plugins: add a tab to the UI, replace a built-in pane, or add a filter command.
• API reference: generated from the source docstrings.