API reference

The public API of Winslow: the names that import winslow exposes.

Workflow

winslow.Workflow

Bases: _ConfigBase

Source code in src/winslow/workflow/workflow.py

class Workflow(_ConfigBase):
    name = None

    # If True, the interactive app initializes this workflow at launch with the
    # default args. It does not show the selector, the form or the confirmation.
    # This is useful for a test. More than one workflow can set it, and the app
    # initializes each of them.
    auto_init = False

    store_classes = {
        Mode.HEADLESS: TaskStore,
        Mode.TUI: InteractiveStore,
    }

    runner_classes = {
        Mode.HEADLESS: HeadlessRunner,
        Mode.TUI: InteractiveRunner,
    }

    graph_class = Graph
    registry_class = TaskRegistry

    filter_registry_class = FilterRegistry

    def __init__(
        self, orchestrator_config, workflow_config=None, store=None, logger=LOGGER
    ):

        super().__init__(orchestrator_config)

        self.workflow_config = (
            workflow_config if workflow_config is not None else Namespace()
        )
        self.registry = self.registry_class(
            orchestrator_config=orchestrator_config,
            workflow_config=self.workflow_config,
        )
        self.graph = self.graph_class(
            orchestrator_config=orchestrator_config,
            workflow_config=self.workflow_config,
        )
        self.filter_registry = self.filter_registry_class(
            orchestrator_config=orchestrator_config,
            workflow_config=self.workflow_config,
        )
        self.logger = logger
        # The Session does not exist at construction time. It is created after
        # the workflow init and attaches itself here.
        self._session = None

        self.batch_options = BatchOptions(
            dry_run=orchestrator_config.dry_run,
            force_run=orchestrator_config.force_run,
            force_success=orchestrator_config.force_success,
            disable_concurrency=orchestrator_config.disable_concurrency,
        )

        if store is None:
            self.logger.debug(f"Auto-initializing store for {self}")
            self.store = self.generate_store(
                orchestrator_config=orchestrator_config,
                workflow_config=self.workflow_config,
            )
        else:
            self.store = store
        self.runner = self.runner_classes[orchestrator_config.mode](
            orchestrator_config=orchestrator_config,
            workflow=self,
            store=self.store,
            logger=self.logger,
            workflow_config=self.workflow_config,
            batch_options=self.batch_options,
        )

    @classmethod
    def generate_store(cls, orchestrator_config, workflow_config):
        return cls.store_classes[orchestrator_config.mode]()

    @property
    def check(self):
        return self.orchestrator_config.check

    @property
    def dry_run(self):
        return self.batch_options.dry_run

    @property
    def force_run(self):
        return self.batch_options.force_run

    @property
    def force_success(self):
        return self.batch_options.force_success

    @property
    def disable_concurrency(self):
        return self.batch_options.disable_concurrency

    @property
    def tasks(self):
        return sorted([t for t in self.store.keys()], key=operator.attrgetter("_index"))

    @property
    def session(self):
        return self._session

    @property
    def session_id(self):
        # A log record can be emitted before a session attaches, so "no session"
        # must be a legal state. The stamp filter accepts None.
        return self._session.session_id if self._session else None

    @property
    def identifier_suffix(self):
        """A key=value list of the config options with identifier=True. This part
        makes two runs of the same workflow different. It is empty if the workflow
        declares no such option. identifier implies required, so each option always
        has a value and the code reads it directly."""
        parts = [
            f"{name}={option.format_value(getattr(self.workflow_config, name))}"
            for name, option in self.config_meta.items()
            if option.identifier
        ]
        return " | ".join(parts)

    @classmethod
    def should_be_initialized(cls, orchestrator_config, parameters=None):
        return True

    def initialize_tasks(self, logger=LOGGER):
        if self.graph is None:
            raise InitializationError(
                f"{self} tasks already initialized - initialize_tasks is one-shot."
            )

        logger.debug(f"{self} initializing tasks.")

        self.registry.collect_classes(self.module_directory)
        tasks = self.graph.generate_pipeline(self.registry)

        for task in tasks:
            self.store[task] = TaskStatus.INITIALIZED

        logger.debug(f"{self} initialized {len(self.store)} tasks.")

        # The graph is necessary only to build the pipeline and to assign the
        # dependencies of each task, which the tasks now hold. Drop the graph, so
        # the garbage collector can free it and its _task_class_map, which holds
        # a reference to each task. The store and the task dependency links then
        # own the tasks.
        self.graph = None

    def release_tasks(self):
        # Drop the references of the workflow to its tasks, so the garbage
        # collector can free each task that nothing else holds. A session calls
        # this when it ends. The store is then the only owner of each task,
        # because the graph is dropped after the init. A clear of the store thus
        # frees each task that an execution-history batch record does not keep,
        # directly or as a dependency of such a record. dict.clear does not take
        # the write lock of the store. This is safe only because the session
        # lifecycle guarantees that no batch runs and that no batch can be
        # admitted here.
        self.store.clear()

    def check_pipeline_eligibility(self, logger=LOGGER):
        tasks = self.tasks
        logger.debug(f"Checking eligibility for {len(tasks)} tasks.")
        self.runner.check_eligibility(tasks)

    @property
    def settings_snapshot(self):
        return {
            "dry_run": self.dry_run,
            "force_run": self.force_run,
            "force_success": self.force_success,
            "check": self.check,
            "disable_concurrency": self.disable_concurrency,
            "env": self.env,
        }

    @property
    def filter(self):
        return getattr(self.orchestrator_config, "filter", None)

    def get_filtered_tasks(self, query=None):
        tasks = self.tasks
        query = query or self.filter
        if not query:
            return tasks
        try:
            return self.filter_registry.parse(query).apply(tasks)
        except ValueError as e:
            # Report the bad filter and do not run everything silently. The parse
            # error message names the exact part that is wrong.
            raise MisconfigurationError(f"Invalid filter: {e}") from e

    def headless_run(self):
        # This looks unused, but the construction of the Session attaches it as
        # _session, and the runner needs it as its logging identity. The
        # interactive path gets its session from the UI, so a headless run must
        # make its own.
        Session(self)

        # Validate and resolve the filter first, so a bad --filter fails
        # immediately, before the eligibility pass runs and logs for each task.
        filtered_tasks = self.get_filtered_tasks()

        if self.filter and not filtered_tasks:
            raise MisconfigurationError(
                f"Filter {self.filter!r} matched no tasks in {self} - nothing to run."
            )

        self.runner.check_eligibility(self.tasks)

        if self.check:
            self.runner.check(filtered_tasks)
        else:
            self.runner.run(filtered_tasks)

        statuses = list(self.store.values())
        completed = sum(1 for s in statuses if s in PASSING_STATUSES)
        unsuccessful = sum(1 for s in statuses if s in UNSUCCESSFUL_STATUSES)

        self.logger.info(
            f"{self} finished - {completed} completed, "
            f"{unsuccessful} unsuccessful, {len(statuses)} total."
        )

        flagged = [
            t for t, s in self.store.items() if s is TaskStatus.COMPLETED_WITH_ERROR
        ]
        if flagged:
            self.logger.warning(
                f"{len(flagged)} task(s) completed despite errors during "
                f"processing - check task logs: {', '.join(map(str, flagged))}"
            )
        return unsuccessful == 0

    @classmethod
    def get_parser(cls, lenient=False):
        if not cls.config_meta:
            return None

        parser = ArgumentParser(
            prog=f"Workflow - {cls.get_name()}",
            description="Sub parser for a workflow",
        )

        for arg_ctx in cls.get_argparse_context(lenient=lenient):
            name = arg_ctx.pop("arg_name")
            parser.add_argument(name, **arg_ctx)

        return parser

identifier_suffix property

A key=value list of the config options with identifier=True. This part makes two runs of the same workflow different. It is empty if the workflow declares no such option. identifier implies required, so each option always has a value and the code reads it directly.

Task

winslow.Task

Bases: _ParameterizationBase

Source code in src/winslow/task/task.py

class Task(_ParameterizationBase):
    class Meta:
        abstract = True

    # The name attribute comes from _Base.
    groups = None

    # Examples of dependency declarations:
    # dependencies = MyTask
    # dependencies = (MyTask1, "MyTask2")
    # dependencies = "MyTask"

    dependencies = None

    # A premier task runs before all tasks that are not premier. A premier task
    # is a foundation step with no explicit dependencies downstream.
    is_premier = False

    # A terminal task runs last, independent of its priority.
    is_terminal = False

    # Tasks in the same priority group are processed concurrently. Set these to
    # False to prevent the run or the completion check from overlapping other
    # work in the batch.
    can_run_parallel = True
    can_check_parallel = True

    # Composable constraints: the Constraint classes that gate the matching
    # hook (see winslow.constraints). A single class or a collection. None
    # means that no constraint is declared. Declare these constraints or
    # override the hook directly. Both run when the gate is evaluated.
    eligibility_constraints = None
    runnability_constraints = None
    checkability_constraints = None
    success_constraints = None
    initialization_constraints = None

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # True when the task defines success. _evaluate_check reads this flag
        # and never calls the default check, which raises NotImplementedError.
        cls._check_overridden = cls.check is not Task.check

    @classmethod
    def _validate_constraint(cls, c, ctype):
        # One validation flow for every constraint that the get_*_constraints
        # methods return, applied at resolve time. The base class follows the
        # lifecycle: an instance hook takes Constraint, and graph init
        # (should_be_initialized) takes ClassConstraint.
        expected = ClassConstraint if ctype in _CLASS_CONSTRAINTS else Constraint
        if isinstance(c, _ConstraintBase):
            raise MisconfigurationError(
                f"{cls.__name__}: {ctype.name} constraint declared as an instance "
                f"({c!r}) - pass the class; the framework initializes it."
            )
        if not (isinstance(c, type) and issubclass(c, expected)):
            raise MisconfigurationError(
                f"{cls.__name__}: {ctype.name} constraint {c!r} must be a "
                f"{expected.__name__} subclass."
            )
        valid_types = c.get_valid_types()
        if valid_types and ctype not in valid_types:
            valid_names = ", ".join(t.name for t in valid_types)
            raise MisconfigurationError(
                f"{cls.__name__}: {c.__name__!r} is not valid as a {ctype.name} "
                f"constraint (valid_types=[{valid_names}])."
            )

    def __init__(self, workflow_config, parameters=None):

        super().__init__(parameters=parameters)

        self.workflow_config = workflow_config

        if self.is_premier and self.is_terminal:
            raise MisconfigurationError(
                f"{self} cannot be declared as premier and terminal at the same time."
            )

        # The graph sets these after it initializes the tasks.
        self._index = None
        self._dependent_tasks = None
        self._priority = None
        self._workflow_name = None

        # All tasks share one logger. The adapter stamps the id of this instance
        # onto each record, so the dispatcher can route the record back to this
        # task. Records propagate to the winslow.runs sink, which stamps
        # task_name.
        self.logger = logging.LoggerAdapter(
            logging.getLogger(TASK_LOGGER_NAME), {"task_id": id(self)}
        )

        self._log_buffer = None

        # Local flag that marks the task as eligible or ineligible. The graph can
        # set it while it assigns the dependencies. The runner then does not
        # check the eligibility again.
        self._is_eligible_result = None

    def _enable_log_buffer(self):
        """Buffer the log records of this task for the Logs tab of the info
        modal. The graph calls this at setup time, in interactive mode only. The
        finalizer drops the buffer when the task is collected. A released task
        therefore frees its buffer, and a task that history retains keeps it."""
        self._log_buffer = collections.deque(maxlen=TASK_LOG_BUFFER_SIZE)
        dispatcher = get_task_dispatcher()
        dispatcher.register_buffer(id(self), self._log_buffer)
        weakref.finalize(self, dispatcher.unregister, id(self))

    @property
    def buffered_logs(self):
        """Buffered log records for the Logs tab of the info modal. None when the
        task is not interactive."""
        return self._log_buffer

    @property
    def _execution_context(self):
        return get_execution_context()

    @property
    def _active_execution_context(self):
        ctx = get_execution_context()
        if ctx is None:
            raise RuntimeError(
                f"Execution flags for {self} are only available during batch "
                f"execution (run/check) - not at import, eligibility or "
                f"graph-build time."
            )
        return ctx

    is_dry_run = _ExecutionFlag("dry_run")
    is_force_run = _ExecutionFlag("force_run")
    is_force_success = _ExecutionFlag("force_success")

    # True after run() of this instance started at least once, in any batch
    # of this workflow. dry_run does not set it. The completion check uses
    # it to separate COMPLETED from COMPLETED_PREVIOUSLY.
    _has_been_run = False

    @property
    def is_noop(self):
        """True for a task with no run action, which only checks completion. The
        value is inferred from run, because the base run does nothing. It thus
        inherits without a declaration: if any class in the MRO overrides run,
        the task is no longer noop."""
        return type(self).run is Task.run

    @property
    def can_process_parallel(self):
        # Processing includes the checks and the run, so it needs both gates.
        # A noop task only checks.
        if self.is_noop:
            return self.can_check_parallel
        return self.can_run_parallel and self.can_check_parallel

    @classmethod
    def get_parameters(cls, workflow_config):

        if not cls._is_parameterized:
            raise ValueError(
                "There is no point trying to get parameters for non-parameterized tasks."
            )

        raise _GetParametersNotImplemented(
            "get_parameters should be explicitly overridden for parametrized tasks to return a"
            " list of dictionaries that match the parameterization fields."
        )

    @classmethod
    def get_groups(cls):
        groups = cls.groups if cls.groups else []
        return frozenset(to_tuple(groups))

    @property
    def groups_readable(self):
        return ", ".join(sorted(self.get_groups()))

    @cached_property
    def _str_cached(self):

        label = str(self.instance_name)

        if self._is_parameterized:
            param_values_readable = ", ".join(
                safe_repr(v) for v in self._parameters_dict.values()
            )

            label = f"{label} ({param_values_readable})"

        return label

    def __str__(self):
        return self._str_cached

    @property
    def dependent_tasks(self):
        return self._dependent_tasks or tuple()

    @property
    def info(self):
        return TaskInfo.from_task(self)

    @classmethod
    def _get_dependency_context(cls):
        """Return the full dependency context of the task in flat form."""
        dependencies = cls.dependencies

        if not dependencies:
            return tuple()

        return flatten(to_tuple(dependencies))

    @property
    def priority(self):
        """A lower priority is processed first. The graph stamps this value with
        the topological generation of the task. This property does not
        calculate it."""
        if self._priority is None:
            raise MisconfigurationError(
                f"{self} priority read before the graph assigned it."
            )
        return self._priority

    @property
    def execution_order(self):
        return (not self.is_premier, self.is_terminal, self.priority)

    @classmethod
    def should_be_initialized(cls, workflow_config, parameters=None):
        """
        Override this to prevent the creation of the task instance.

        This is stronger than is_eligible, which filters the tasks after their
        creation. Use it with task parameterization when most of the
        parameterized tasks are skipped.

        A task that is not initialized does not show on the UI. This can confuse
        a user who declared the task.
        """
        return True

    @classmethod
    def get_initialization_constraints(cls, workflow_config):
        """Override this to select should_be_initialized constraints by env or by
        config."""
        return cls.initialization_constraints

    @classmethod
    def _evaluate_should_be_initialized(cls, workflow_config, parameters=None):
        # Class-level constraints take (task_class, parameters). The get_* method
        # supplies them. They go through the same validate and resolve flow, and
        # they run with the override.
        for c in to_tuple(cls.get_initialization_constraints(workflow_config) or ()):
            if not cls._prepare_constraint(
                c, ConstraintType.INITIALIZATION, workflow_config
            )(cls, parameters):
                return False
        return cls.should_be_initialized(workflow_config, parameters=parameters)

    def depends_on(self, task):
        """
        Override this to control if a task instance depends on another task
        instance.

        Use it to prevent a cyclic dependency between two task classes that
        depend on each other. Parameterization makes this more frequent.
        """
        return True

    def is_eligible(self):
        """Override this to control the eligibility of the task for the workflow."""
        return True

    def can_run(self):
        """Override this to block a task that does not satisfy your constraints."""
        return True

    def can_check(self):
        """Override this to gate the completion check. Return False, or call
        self.block, to mark the task BLOCKED and not run check. Use it when the
        check depends on data that is not always available."""
        return True

    def check(self):
        """Override this to implement the success check."""
        raise NotImplementedError

    # -- constraint evaluation -------------------------------------------------
    # The runner calls the sealed _evaluate_* methods and never the hooks. The
    # declared constraints thus always run with an override. The constraints run
    # first, so a cheap declarative guard can stop the evaluation before the
    # custom logic.

    # The get_*_constraints methods return the constraints for each gate. The
    # default is the declared class attribute. Override them to select the
    # constraints by env or by self.workflow_config.
    def get_eligibility_constraints(self):
        return self.eligibility_constraints

    def get_runnability_constraints(self):
        return self.runnability_constraints

    def get_checkability_constraints(self):
        return self.checkability_constraints

    def get_success_constraints(self):
        return self.success_constraints

    @cached_property
    def _constraint_instances(self):
        """Resolve the instance-level constraints one time, from the
        get_*_constraints methods. Validate each constraint, then instantiate
        it."""
        return {
            ctype: tuple(
                self._prepare_constraint(c, ctype, self.workflow_config)
                for c in to_tuple(getattr(self, f"get_{attr}")() or ())
            )
            for ctype, attr in _INSTANCE_CONSTRAINTS.items()
        }

    @classmethod
    def _prepare_constraint(cls, c, ctype, workflow_config):
        """Validate the constraint, then instantiate it. Every constraint from
        get_*_constraints goes through this one step."""
        cls._validate_constraint(c, ctype)
        return c(workflow_config)

    def _check_constraints(self, ctype):
        return all(c(self) for c in self._constraint_instances[ctype])

    def _evaluate_is_eligible(self):
        return (
            self._check_constraints(ConstraintType.ELIGIBILITY) and self.is_eligible()
        )

    def _evaluate_can_run(self):
        return self._check_constraints(ConstraintType.RUNNABILITY) and self.can_run()

    def _evaluate_can_check(self):
        return self._check_constraints(ConstraintType.CHECKABILITY) and self.can_check()

    def _evaluate_check(self):
        constraints = self._constraint_instances[ConstraintType.SUCCESS]
        # The task must define success. If it does not, all([]) reports completed
        # with no check. This runs here and not at import, because
        # get_success_constraints supplies the constraints and can depend on the
        # env.
        if not self._check_overridden and not constraints:
            raise MisconfigurationError(
                f"{type(self).__name__} must define success: override check() "
                f"or provide success_constraints."
            )
        result = all(c(self) for c in constraints)
        if self._check_overridden:
            result = result and self.check()
        return result

    def run(self):
        """Make the system change. The check method then confirms the change."""
        pass

    def dry_run(self):
        """The runner calls this instead of run in dry-run mode. Override it to
        simulate the change. The default makes no change."""
        pass

    def block(self, msg):
        raise exceptions.TaskBlock(msg)

    def skip(self, msg):
        raise exceptions.TaskSkip(msg)

    def fail(self, msg):
        raise exceptions.TaskFailure(msg)

    def require_action(self, msg):
        raise TaskActionRequired(msg)

    def _check_and_raise(self, condition, msg, method):
        result = condition() if callable(condition) else condition

        if result:
            method(msg)

    def block_if(self, condition, msg):
        self._check_and_raise(condition, msg, self.block)

    def skip_if(self, condition, msg):
        self._check_and_raise(condition, msg, self.skip)

    def fail_if(self, condition, msg):
        self._check_and_raise(condition, msg, self.fail)

buffered_logs property

Buffered log records for the Logs tab of the info modal. None when the task is not interactive.

is_noop property

True for a task with no run action, which only checks completion. The value is inferred from run, because the base run does nothing. It thus inherits without a declaration: if any class in the MRO overrides run, the task is no longer noop.

priority property

A lower priority is processed first. The graph stamps this value with the topological generation of the task. This property does not calculate it.

should_be_initialized(workflow_config, parameters=None) classmethod

Override this to prevent the creation of the task instance.

This is stronger than is_eligible, which filters the tasks after their creation. Use it with task parameterization when most of the parameterized tasks are skipped.

A task that is not initialized does not show on the UI. This can confuse a user who declared the task.

Source code in src/winslow/task/task.py

@classmethod
def should_be_initialized(cls, workflow_config, parameters=None):
    """
    Override this to prevent the creation of the task instance.

    This is stronger than is_eligible, which filters the tasks after their
    creation. Use it with task parameterization when most of the
    parameterized tasks are skipped.

    A task that is not initialized does not show on the UI. This can confuse
    a user who declared the task.
    """
    return True

get_initialization_constraints(workflow_config) classmethod

Override this to select should_be_initialized constraints by env or by config.

Source code in src/winslow/task/task.py

@classmethod
def get_initialization_constraints(cls, workflow_config):
    """Override this to select should_be_initialized constraints by env or by
    config."""
    return cls.initialization_constraints

depends_on(task)

Override this to control if a task instance depends on another task instance.

Use it to prevent a cyclic dependency between two task classes that depend on each other. Parameterization makes this more frequent.

Source code in src/winslow/task/task.py

def depends_on(self, task):
    """
    Override this to control if a task instance depends on another task
    instance.

    Use it to prevent a cyclic dependency between two task classes that
    depend on each other. Parameterization makes this more frequent.
    """
    return True

is_eligible()

Override this to control the eligibility of the task for the workflow.

Source code in src/winslow/task/task.py

def is_eligible(self):
    """Override this to control the eligibility of the task for the workflow."""
    return True

can_run()

Override this to block a task that does not satisfy your constraints.

Source code in src/winslow/task/task.py

def can_run(self):
    """Override this to block a task that does not satisfy your constraints."""
    return True

can_check()

Override this to gate the completion check. Return False, or call self.block, to mark the task BLOCKED and not run check. Use it when the check depends on data that is not always available.

Source code in src/winslow/task/task.py

def can_check(self):
    """Override this to gate the completion check. Return False, or call
    self.block, to mark the task BLOCKED and not run check. Use it when the
    check depends on data that is not always available."""
    return True

check()

Override this to implement the success check.

Source code in src/winslow/task/task.py

def check(self):
    """Override this to implement the success check."""
    raise NotImplementedError

run()

Make the system change. The check method then confirms the change.

Source code in src/winslow/task/task.py

def run(self):
    """Make the system change. The check method then confirms the change."""
    pass

dry_run()

The runner calls this instead of run in dry-run mode. Override it to simulate the change. The default makes no change.

Source code in src/winslow/task/task.py

def dry_run(self):
    """The runner calls this instead of run in dry-run mode. Override it to
    simulate the change. The default makes no change."""
    pass

TaskFilter

winslow.TaskFilter

Bases: Registerable

Source code in src/winslow/filter/base.py

class TaskFilter(Registerable):
    short_command = None
    long_command = None

    def __init__(self, value):
        self.value = value

    @classmethod
    @lru_cache
    def get_name(cls):
        # Use long_command first, then short_command, then the kebab class name.
        return cls.long_command or cls.short_command or super().get_name()

    def matches(self, task) -> bool:
        raise NotImplementedError

    def explain(self) -> str:
        raise NotImplementedError

Constraints

winslow.Constraint

Bases: _ConstraintBase

Constraint for the instance hooks: is_eligible, can_run, can_check and check. It is called with the live task.

Source code in src/winslow/constraints.py

class Constraint(_ConstraintBase):
    """Constraint for the instance hooks: is_eligible, can_run,
    can_check and check. It is called with the live task."""

    def __call__(self, task):
        return self.apply(task)

    def apply(self, task):
        raise NotImplementedError

winslow.ClassConstraint

Bases: _ConstraintBase

Constraint for should_be_initialized. It is evaluated at graph init, before an instance exists, and is called with the task class and the parameters.

Source code in src/winslow/constraints.py

class ClassConstraint(_ConstraintBase):
    """Constraint for should_be_initialized. It is evaluated at graph init, before
    an instance exists, and is called with the task class and the parameters."""

    def __call__(self, task_class, parameters=None):
        return self.apply(task_class, parameters)

    def apply(self, task_class, parameters=None):
        raise NotImplementedError

winslow.ConstraintType

Bases: Enum

Source code in src/winslow/constraints.py

class ConstraintType(Enum):
    # The task gate for each constraint list. INITIALIZATION is evaluated at
    # graph init, on the class (should_be_initialized), so its constraints take
    # (task_class, parameters). Each other type gates a live instance and takes
    # (task). CHECKABILITY gates the completion check and blocks the task if it
    # fails. SUCCESS adds to the success predicate (check).
    INITIALIZATION = auto()
    ELIGIBILITY = auto()
    RUNNABILITY = auto()
    CHECKABILITY = auto()
    SUCCESS = auto()

Descriptors

winslow.Parameter dataclass

Declare a parameter on a Task class. An instance is a Python descriptor. Access on the class returns the Parameter. Access on an instance returns the resolved value from task._params.

One Parameter usually binds one task attribute. from_tuple and from_dict declare a compound parameter, which binds more than one attribute. See those classmethods. name binds the value to an attribute with a different name than the declared one.

Source code in src/winslow/descriptors.py

@dataclass
class Parameter:
    """
    Declare a parameter on a Task class. An instance is a Python descriptor.
    Access on the class returns the Parameter. Access on an instance returns the
    resolved value from task._params.

    One Parameter usually binds one task attribute. `from_tuple` and `from_dict`
    declare a compound parameter, which binds more than one attribute. See those
    classmethods. `name` binds the value to an attribute with a different name
    than the declared one.
    """

    values: Union[Callable, Any, None] = None
    allowed_types: Optional[Tuple[type, ...]] = None
    param_style: ParameterStyle = ParameterStyle.SEQUENTIAL
    name: Optional[str] = None

    # An InitVar, to process the raw initialization value.
    _raw_values: InitVar[Union[Callable, Any, None]] = None

    # Markers for a compound parameter (from_tuple or from_dict). These are plain
    # class attributes and not fields.
    # _compound: this Parameter expands into more than one attribute.
    # _compound_kind: "tuple" or "dict".
    # _compound_names: the explicit attribute names from names= in from_tuple, or
    # None to derive them from the declared name.
    # _compound_name_map: the {dict-key: attribute-name} translation from
    # from_dict, or None.
    _compound = False
    _compound_kind = None
    _compound_names = None
    _compound_name_map = None

    def __post_init__(self, _raw_values):
        if _raw_values is None:
            _raw_values = self.values

        if (
            _raw_values is not None
            and not callable(_raw_values)
            and self.allowed_types is not None
        ):
            for value in to_tuple(_raw_values):
                if not isinstance(value, self.allowed_types):
                    raise ParameterizationError(
                        f"{value} is not of type: {self.allowed_types}."
                    )

    @classmethod
    def from_tuple(cls, values, names=None):
        """Declare more than one attribute from rows of aligned values.

        `foo_bar = Parameter.from_tuple([(1, 2), (3, 4)])` gives two tasks:
        `(foo=1, bar=2)` and `(foo=3, bar=4)`. The attribute names come from the
        declared name, which is split on `_`, or from `names=("foo", "bar")`.
        `values` is a list of rows, or a callable(workflow_config) that returns
        one. The rows are aligned, so each row is one combination. They combine
        with each other parameter as an independent cartesian axis.
        """
        param = cls(values=values)
        param._compound = True
        param._compound_kind = "tuple"
        param._compound_names = tuple(names) if names is not None else None
        return param

    @classmethod
    def from_dict(cls, values, name_map=None):
        """Declare more than one attribute from rows of dicts.

        `foo_bar = Parameter.from_dict([{"foo": 1, "bar": 2}, {"foo": 3, "bar": 4}])`
        gives two tasks: `(foo=1, bar=2)` and `(foo=3, bar=4)`. The attribute
        names come from the declared name, as in `from_tuple`. Each row must have
        the same keys. If the dict keys are different from the attribute names,
        pass `name_map={dict_key: attribute_name}` to translate them. `values` is
        a list of dicts, or a callable(workflow_config) that returns one. The rows
        combine with each other parameter as an independent cartesian axis.
        """
        param = cls(values=values)
        param._compound = True
        param._compound_kind = "dict"
        param._compound_name_map = dict(name_map) if name_map is not None else None
        return param

    def __set_name__(self, owner, name):
        self._attr_name = name

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj._params, self._attr_name)

    def __set__(self, obj, value):
        raise ParameterizationError(
            f"'{self._attr_name}' is a parameter and cannot be assigned."
        )

from_tuple(values, names=None) classmethod

Declare more than one attribute from rows of aligned values.

foo_bar = Parameter.from_tuple([(1, 2), (3, 4)]) gives two tasks: (foo=1, bar=2) and (foo=3, bar=4). The attribute names come from the declared name, which is split on _, or from names=("foo", "bar"). values is a list of rows, or a callable(workflow_config) that returns one. The rows are aligned, so each row is one combination. They combine with each other parameter as an independent cartesian axis.

Source code in src/winslow/descriptors.py

@classmethod
def from_tuple(cls, values, names=None):
    """Declare more than one attribute from rows of aligned values.

    `foo_bar = Parameter.from_tuple([(1, 2), (3, 4)])` gives two tasks:
    `(foo=1, bar=2)` and `(foo=3, bar=4)`. The attribute names come from the
    declared name, which is split on `_`, or from `names=("foo", "bar")`.
    `values` is a list of rows, or a callable(workflow_config) that returns
    one. The rows are aligned, so each row is one combination. They combine
    with each other parameter as an independent cartesian axis.
    """
    param = cls(values=values)
    param._compound = True
    param._compound_kind = "tuple"
    param._compound_names = tuple(names) if names is not None else None
    return param

from_dict(values, name_map=None) classmethod

Declare more than one attribute from rows of dicts.

foo_bar = Parameter.from_dict([{"foo": 1, "bar": 2}, {"foo": 3, "bar": 4}]) gives two tasks: (foo=1, bar=2) and (foo=3, bar=4). The attribute names come from the declared name, as in from_tuple. Each row must have the same keys. If the dict keys are different from the attribute names, pass name_map={dict_key: attribute_name} to translate them. values is a list of dicts, or a callable(workflow_config) that returns one. The rows combine with each other parameter as an independent cartesian axis.

Source code in src/winslow/descriptors.py

@classmethod
def from_dict(cls, values, name_map=None):
    """Declare more than one attribute from rows of dicts.

    `foo_bar = Parameter.from_dict([{"foo": 1, "bar": 2}, {"foo": 3, "bar": 4}])`
    gives two tasks: `(foo=1, bar=2)` and `(foo=3, bar=4)`. The attribute
    names come from the declared name, as in `from_tuple`. Each row must have
    the same keys. If the dict keys are different from the attribute names,
    pass `name_map={dict_key: attribute_name}` to translate them. `values` is
    a list of dicts, or a callable(workflow_config) that returns one. The rows
    combine with each other parameter as an independent cartesian axis.
    """
    param = cls(values=values)
    param._compound = True
    param._compound_kind = "dict"
    param._compound_name_map = dict(name_map) if name_map is not None else None
    return param

winslow.ConfigOption dataclass

Source code in src/winslow/descriptors.py

@dataclass
class ConfigOption:
    type: Optional[Type] = None
    default: Optional[Any] = None
    # The inverse of `type`. str() cannot invert it for a value that is not a
    # scalar: a list default becomes "[1]", which does not parse again. An option
    # can thus supply its own serializer from a value to a string.
    serializer: Optional[Callable] = None
    help_text: str = ""
    required: bool = False
    choices: Optional[List[Any]] = None
    multiselect: bool = False
    action: Optional[str] = None
    const: Optional[Any] = None  # used only if action == "store_const"
    show_on_ui: bool = True
    # Part of the display label of the instance. It implies required.
    identifier: bool = False
    subcommands: Tuple[Any, ...] = field(default_factory=tuple)
    depends_on: Tuple[str, ...] = field(default_factory=tuple)
    name: Optional[str] = None

    def __post_init__(self):
        self.subcommands = to_tuple(self.subcommands) if self.subcommands else tuple()
        self.depends_on = to_tuple(self.depends_on) if self.depends_on else tuple()
        if self.identifier:
            # An identifier with no value distinguishes nothing, so the user must
            # supply it.
            self.required = True

    def format_value(self, value):
        if value is None:
            return None
        if self.serializer:
            return self.serializer(value)
        if isinstance(value, (list, tuple)):
            # A multiselect value is a list. str() of a list is noisy in a label.
            return ", ".join(str(v) for v in value)
        return str(value)

    def to_arg(self, name=None, lenient=False):
        """Make the argparse argument definition. `lenient` removes `required`, so
        a parser that is shared between workflows does not abort on a mandatory
        option of one workflow. The strict parse of a single workflow enforces the
        value."""
        name = (name or self.name).replace("_", "-")

        common = dict(
            arg_name=f"--{name}",
            help=self.help_text,
            # A default supplies the value, so the command line does not demand
            # the option again.
            required=self.required and self.default is None and not lenient,
            default=self.default,
        )

        if self.action:
            if self.action == "store_const":
                return dict(common, action=self.action, const=self.const)
            else:
                return dict(common, action=self.action)

        result = dict(
            common,
            type=self.type,
            choices=self.choices,
        )
        if self.multiselect:
            # The form takes more than one value, so the command line must
            # take them too.
            result["nargs"] = "+"
        return result

to_arg(name=None, lenient=False)

Make the argparse argument definition. lenient removes required, so a parser that is shared between workflows does not abort on a mandatory option of one workflow. The strict parse of a single workflow enforces the value.

Source code in src/winslow/descriptors.py

def to_arg(self, name=None, lenient=False):
    """Make the argparse argument definition. `lenient` removes `required`, so
    a parser that is shared between workflows does not abort on a mandatory
    option of one workflow. The strict parse of a single workflow enforces the
    value."""
    name = (name or self.name).replace("_", "-")

    common = dict(
        arg_name=f"--{name}",
        help=self.help_text,
        # A default supplies the value, so the command line does not demand
        # the option again.
        required=self.required and self.default is None and not lenient,
        default=self.default,
    )

    if self.action:
        if self.action == "store_const":
            return dict(common, action=self.action, const=self.const)
        else:
            return dict(common, action=self.action)

    result = dict(
        common,
        type=self.type,
        choices=self.choices,
    )
    if self.multiselect:
        # The form takes more than one value, so the command line must
        # take them too.
        result["nargs"] = "+"
    return result

Decorators

winslow.decorators.transient_property = TransientProperty module-attribute

Orchestrator

winslow.Orchestrator

Bases: _ConfigBase

Source code in src/winslow/orchestrator.py

class Orchestrator(_ConfigBase):
    workflow_registry_class = WorkflowRegistry

    workflow = ConfigOption(
        help_text="Name of the workflow to view / run.",
        required=False,
        subcommands=(
            Action.RUN.value,
            Action.SHOW.value,
        ),
        # The UI has a workflow selection widget, so it needs no text input.
        show_on_ui=False,
    )

    filter = ConfigOption(
        help_text=(
            "Filter expression for tasks. Supports name matching (bare text), "
            "filter commands (!g <group>, !group <group>), "
            "boolean operators (& |), negation (~), grouping (()), "
            "and comma-separated OR shorthand (foo,bar)."
        ),
        required=False,
        subcommands=(
            Action.RUN.value,
            Action.SHOW.value,
        ),
        depends_on="initialize",
        show_on_ui=False,
    )

    initialize = ConfigOption(
        action="store_true",
        required=False,
        default=False,
        help_text=(
            "Initialize a single workflow (chosen with --workflow) and list its "
            "tasks. Required in order to use --filter with show."
        ),
        subcommands=Action.SHOW.value,
        show_on_ui=False,
    )

    with_deps = ConfigOption(
        action="store_true",
        required=False,
        default=False,
        help_text=(
            "With --initialize, also list each task's dependencies, in the order "
            "they run."
        ),
        subcommands=Action.SHOW.value,
        depends_on="initialize",
        show_on_ui=False,
    )

    mode = ConfigOption(
        type=_parse_mode,
        choices=tuple(Mode),
        required=False,
        help_text=(
            "How to run the workflow(s): 'tui' launches the interactive terminal"
            " UI; 'headless' runs single-thread/single-process with no UI, useful"
            " for CI and debugging."
        ),
        default=Mode.TUI,
        subcommands=Action.RUN.value,
        show_on_ui=False,
    )

    check = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Checks the tasks in the workflow instead of running them"
            " - available in headless run mode."
        ),
        default=False,
        subcommands=Action.RUN.value,
        show_on_ui=False,
    )

    disable_concurrency = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Disables all concurrency during task runs (e.g. task dependencies and task "
            "eligibility will be checked sequentially)."
        ),
        default=False,
        subcommands=Action.RUN.value,
    )

    dry_run = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Run tasks in dry-run mode (dry_run method will be "
            "called on tasks, instead of run)"
        ),
        default=False,
        subcommands=Action.RUN.value,
    )

    force_run = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Skips implicit success checks before a task run. "
            "Runnability and dependency checks will still be in effect."
        ),
        default=False,
        subcommands=Action.RUN.value,
    )

    force_success = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Mark tasks as successful (FORCE_SUCCESS) without running any checks "
            "or the task itself. Ineligible (skipped) tasks are left untouched."
        ),
        default=False,
        subcommands=Action.RUN.value,
    )

    reraise_errors = ConfigOption(
        action="store_true",
        required=False,
        help_text=(
            "Re-raise unexpected task errors after marking the task ERROR, "
            "aborting the run instead of continuing. For CI and debugging."
        ),
        default=False,
        subcommands=Action.RUN.value,
        show_on_ui=False,
    )

    debug = ConfigOption(
        action="store_true",
        required=False,
        help_text="Enable debug mode and logging.",
        default=False,
        subcommands=(
            Action.RUN.value,
            Action.SHOW.value,
        ),
        # The command line enables the debug mode.
        show_on_ui=False,
    )

    def __init__(self, orchestrator_config, directory=None, unknown_args=None):
        self.directory = directory or os.getcwd()
        # argparse reads sys.argv if this is None. Normalize it at the boundary.
        self.unknown_args = list(unknown_args or ())

        super().__init__(orchestrator_config)

        self.workflow_registry = self.workflow_registry_class(self.orchestrator_config)

        # Set later if the interactive mode is enabled.
        self.ui = None

        self.logger = LOGGER

    @property
    def is_interactive(self):
        return self.orchestrator_config.is_interactive

    @classmethod
    def _add_arguments(cls, parser, subcommand=None):
        for arg_ctx in cls.get_argparse_context(subcommand):
            try:
                name = arg_ctx.pop("arg_name")
                parser.add_argument(name, **arg_ctx)
            except (TypeError, KeyError):
                cmd = subcommand if subcommand else "base"
                LOGGER.error(
                    f"Could not add argument for {cmd} parser {parser}, context: '{name}' - {arg_ctx}"
                )
                raise

    @classmethod
    def _generate_subcommand(
        cls,
        subparsers,
        action,
        help_text,
    ):
        sub_parser = subparsers.add_parser(action.value, help=help_text)

        cls._add_arguments(sub_parser, subcommand=action.value)
        sub_parser.set_defaults(action=action)

        return sub_parser

    @classmethod
    def get_base_parser(cls):

        parser = argparse.ArgumentParser(
            prog="Winslow",
            description="A state and workflow management framework with a terminal UI",
        )

        cls._add_arguments(parser)

        subparsers = parser.add_subparsers(help="Available subcommands")

        all_defaults = {
            name: conf.default
            for name, conf in cls.config_meta.items()
            if conf.subcommands
        }
        parser.set_defaults(action=Action.RUN, **all_defaults)

        cls._generate_subcommand(
            subparsers,
            action=Action.SHOW,
            help_text="Show workflow / task information.",
        )

        cls._generate_subcommand(
            subparsers, action=Action.RUN, help_text="Run workflow(s)."
        )

        return parser

    @classmethod
    def parse_args(cls):
        base_parser = cls.get_base_parser()
        base_args, unknown_args = base_parser.parse_known_args(
            namespace=OrchestratorConfig()
        )
        return base_args, unknown_args

    @property
    def sorted_workflow_classes(self):
        return sorted(
            self.workflow_registry.classes,
            key=lambda kls: (kls.get_module_path(), kls.__name__),
        )

    @classmethod
    def get_from_cwd(cls):
        klasses = []
        for scoped_name in iter_dir_module_names(os.getcwd(), recursive=False):
            try:
                module = importlib.import_module(scoped_name)
            # Catch SystemExit too. A sys.exit() in an unguarded script must not
            # stop the CLI.
            except (Exception, SystemExit) as e:
                # The orchestrator is optional. A broken file that has no
                # relation to it must not stop the CLI.
                LOGGER.warning(
                    f"Skipping {scoped_name.split('.', 1)[1]}.py - "
                    f"{type(e).__name__}: {e}",
                    exc_info=True,
                )
                continue
            klasses.extend(classes_in_module(module, cls))

        if len(klasses) > 1:
            names = ", ".join(kls.__name__ for kls in klasses)

            raise MisconfigurationError(
                f"Multiple concrete orchestrator classes found in {os.getcwd()}: {names}. "
                f"Please define one orchestrator at most per directory."
            )

        return klasses[0] if klasses else None

    def _parse_known_workflow_args(self, workflow_kls, unknown):
        """Returns the parsed args, or None, and the tokens that this workflow did
        not claim."""
        parser = workflow_kls.get_parser(lenient=True)
        if parser is None:
            return None, tuple(unknown)
        try:
            # Lenient: the args of another workflow are not for this parser.
            args, leftover = parser.parse_known_args(unknown)
        except SystemExit:
            # argparse already printed the message. Send the exit to the clean
            # error path.
            raise MisconfigurationError(
                f"Invalid arguments for workflow '{workflow_kls.get_name()}' (see above)."
            )
        return args, tuple(leftover)

    @classmethod
    def _leftover_indices(cls, unknown, leftover):
        """The positions in `unknown` that argparse did NOT claim for this
        workflow.

        The positions are matched, and not the values. This is important.

        Two workflows each take a number, and the user runs:

            --my-integer 7 --batch-size 7

        Workflow A claims `--my-integer 7`. Workflow B claims `--batch-size 7`.
        Each token is claimed, so the user must get no error.

        A comparison by value makes the two `7` tokens equal: the leftover of A
        holds a `7`, which belongs to B, and the leftover of B holds a `7`, which
        belongs to A. The value `7` thus looks unclaimed and the result is the
        wrong message "Unrecognized arguments: 7". The positions keep the two
        tokens separate: the `7` of A is at position 1 and the `7` of B is at
        position 3.

            unknown = ['--my-integer', '7', '--batch-size', '7']
            positions:  0              1     2               3

        argparse keeps the leftovers in order. This method thus walks `unknown`,
        matches the leftovers against it, and finds the positions that this
        workflow did not claim."""
        positions = set()
        cursor = 0
        for i, token in enumerate(unknown):
            if cursor < len(leftover) and token == leftover[cursor]:
                positions.add(i)
                cursor += 1
        return positions

    def _collect_workflow_args(self):
        """{workflow_kls: parsed args or None}, validated together.

        Only `show` and the interactive start call this. They parse the args of
        each discovered workflow at one time. A headless run has one named
        workflow and parses its args strictly (see `_handle_headless_run`), so
        the collision between workflows that this method handles cannot occur
        there."""
        parsed = {
            kls: self._parse_known_workflow_args(kls, self.unknown_args)
            for kls in self.sorted_workflow_classes
        }

        # A token is unrecognized if no workflow claimed its position. A position
        # that one workflow claimed leaves the intersection.
        unclaimed = set(range(len(self.unknown_args)))
        for _, leftover in parsed.values():
            unclaimed &= self._leftover_indices(self.unknown_args, leftover)

        if unclaimed:
            tokens = [self.unknown_args[i] for i in sorted(unclaimed)]
            raise MisconfigurationError(f"Unrecognized arguments: {' '.join(tokens)}")

        return {kls: args for kls, (args, _) in parsed.items()}

    def _display_workflow(self, idx, workflow, show_tasks):
        self.logger.info(
            f"{idx + 1}. {workflow.__class__.__name__} ({workflow.instance_name})"
            f" - {workflow.module_directory}"
        )

        if show_tasks:
            self._display_tasks(workflow)
        else:
            self._display_task_classes(workflow)

    def _display_task_classes(self, workflow):
        for i, kls in enumerate(
            sorted(workflow.registry.classes, key=lambda k: k.get_name())
        ):
            parameterized = getattr(kls, "_is_parameterized", False)
            marker = "  [parameterized]" if parameterized else ""
            self.logger.info(f"{INDENT}{i + 1}. {kls.__name__}{marker}")

            if parameterized:
                for declared_name, param in kls._parameterization_meta.items():
                    self.logger.info(
                        f"{INDENT * 2}- {self._describe_parameter(declared_name, param)}"
                    )

    @classmethod
    def _describe_parameter(cls, declared_name, param):
        style = param.param_style.name.lower()
        resolved = getattr(param, "_resolved_names", None)
        if resolved:
            return f"{declared_name} -> {', '.join(resolved)}  ({style})"
        return f"{declared_name}  ({style})"

    def _display_tasks(self, workflow):
        for task in workflow.get_filtered_tasks():
            self.logger.info(f"{INDENT}{task._index + 1}. {str(task)}")

            if self.orchestrator_config.with_deps and task.dependent_tasks:
                self.logger.info(f"{INDENT * 2}Dependencies:")
                for dep in sorted(task.dependent_tasks, key=lambda d: d._index):
                    self.logger.info(f"{INDENT * 3}{dep._index + 1}. {str(dep)}")

    def check_filters(self, workflow):
        wanted = getattr(self.orchestrator_config, "workflow", None)
        return not wanted or workflow.get_name() == wanted

    def _handle_show(self):
        if self.orchestrator_config.initialize:
            return self._handle_initialize()
        self._list_workflow_classes()

    def _list_workflow_classes(self):
        self.logger.debug("Listing workflow classes")
        workflow_args_map = self._collect_workflow_args()
        workflows = []

        for workflow_kls in self.sorted_workflow_classes:
            if not workflow_kls.should_be_initialized(self.orchestrator_config):
                continue

            workflow = workflow_kls(
                self.orchestrator_config, workflow_args_map[workflow_kls]
            )

            if not self.check_filters(workflow):
                continue

            workflow.registry.collect_classes(workflow.module_directory)
            workflows.append(workflow)

        for idx, workflow in enumerate(workflows):
            self._display_workflow(idx, workflow, show_tasks=False)

    def _handle_initialize(self):
        self.logger.debug("Initializing a single workflow")
        workflow_name = self.orchestrator_config.workflow

        if not workflow_name:
            raise MisconfigurationError(
                "--initialize requires a single workflow - pass --workflow <name>."
            )

        if workflow_name not in self.workflow_registry:
            raise MisconfigurationError(
                f"{workflow_name} not found in the workflow registry"
                f" - (available workflows: {self.workflow_registry.names})"
            )

        workflow_kls = self.workflow_registry[workflow_name]
        workflow_args = self._parse_workflow_args(workflow_kls)

        if not workflow_kls.should_be_initialized(self.orchestrator_config):
            raise InitializationError(
                f"Cannot initialize workflow {workflow_kls} - should_be_initialized check failed."
            )

        workflow = workflow_kls(self.orchestrator_config, workflow_args)
        workflow.initialize_tasks()
        self._display_workflow(0, workflow, show_tasks=True)

    def _parse_workflow_args(self, workflow_kls):
        """Parse strictly for the one workflow that is selected to run. `required`
        and depends_on are thus enforced here. The lenient collective parse, which
        only lists the workflows, does not enforce them."""
        parser = workflow_kls.get_parser()
        if parser is None:
            if self.unknown_args:
                raise MisconfigurationError(
                    f"Unrecognized arguments: {' '.join(self.unknown_args)}"
                )
            return None
        args = parser.parse_args(self.unknown_args)
        workflow_kls.validate_option_dependencies(args)
        return args

    def _handle_run(self):
        if self.orchestrator_config.mode is Mode.HEADLESS:
            return self._handle_headless_run()
        self._handle_interactive_run()

    def _handle_headless_run(self):
        self.logger.debug("Headless run")

        workflow_name = self.orchestrator_config.workflow

        if not workflow_name:
            raise MisconfigurationError(
                "Need to provide workflow name via --workflow parameter."
            )

        if workflow_name not in self.workflow_registry:
            raise MisconfigurationError(
                f"{workflow_name} not found in the workflow registry"
                f" - (available workflows: {self.workflow_registry.names})"
            )

        workflow_kls = self.workflow_registry[workflow_name]
        workflow_args = self._parse_workflow_args(workflow_kls)

        if not workflow_kls.should_be_initialized(self.orchestrator_config):
            raise InitializationError(
                f"Cannot initialize workflow {workflow_kls} - should_be_initialized check failed."
            )

        workflow = workflow_kls(self.orchestrator_config, workflow_args)
        workflow.initialize_tasks()
        return workflow.headless_run()

    def _handle_interactive_run(self):
        self.logger.debug("Interactive run")

        try:
            from winslow.ui import Winslow
        except ImportError as e:
            raise MisconfigurationError(
                "Interactive mode requires the UI extra - install with: pip install 'winslow[tui]'"
            ) from e

        # Validate the CLI args before any effect, so a typo fails immediately.
        workflow_args_map = self._collect_workflow_args()

        # Set up the winslow.runs sink and the propagate=False boundary BEFORE a
        # workflow logger or a task logger starts to propagate. The run logs then
        # go to the sink and not to the console. This is interactive only. A
        # headless run keeps the console output.
        setup_run_logging()

        # The parsed workflow args fill the parameter forms of the UI.
        workflow_context = {
            kls: args
            for kls, args in workflow_args_map.items()
            if kls.should_be_initialized(self.orchestrator_config)
        }

        self.app = Winslow(
            orchestrator_config=self.orchestrator_config,
            orchestrator=self,
            workflow_context=workflow_context,
        )

        # app.run() blocks until the TUI stops. Then flush and stop the
        # run-logging listener. This is in a finally clause, so it also occurs
        # after an error. If it does not occur, the listener thread and the open
        # file handles stay, and the buffered lines are lost.
        try:
            self.app.run()
        finally:
            shutdown_run_logging()

    def initialize_workflow(
        self,
        workflow_kls,
        orchestrator_overrides,
        workflow_values,
        workflow_base=None,
        task_store=None,
        logger=LOGGER,
    ):
        # Put the UI values on the parsed base and do not build a new config from
        # them. The base holds the default of each declared option, and also of
        # an option with show_on_ui=False. The form path thus cannot drop an
        # option that it did not show.
        orchestrator_config = _merged_config(
            self.orchestrator_config, orchestrator_overrides
        )
        workflow_params = _merged_config(workflow_base, workflow_values)

        return workflow_kls(
            orchestrator_config, workflow_params, store=task_store, logger=logger
        )

    def start(self):
        """
        Do the action that self.orchestrator_config (base_args) and unknown_args
        select:

        - Start the UI.
        - Start a simple run.
        - Show a summary of the workflows and the tasks.
        """
        self.validate_option_dependencies(
            self.orchestrator_config, subcommand=self.orchestrator_config.action.value
        )

        self.workflow_registry.collect_classes(self.directory)

        if self.orchestrator_config.action is Action.SHOW:
            self._handle_show()
        elif self.orchestrator_config.action is Action.RUN:
            return self._handle_run()

start()

Do the action that self.orchestrator_config (base_args) and unknown_args select:

• Start the UI.
• Start a simple run.
• Show a summary of the workflows and the tasks.

Source code in src/winslow/orchestrator.py

def start(self):
    """
    Do the action that self.orchestrator_config (base_args) and unknown_args
    select:

    - Start the UI.
    - Start a simple run.
    - Show a summary of the workflows and the tasks.
    """
    self.validate_option_dependencies(
        self.orchestrator_config, subcommand=self.orchestrator_config.action.value
    )

    self.workflow_registry.collect_classes(self.directory)

    if self.orchestrator_config.action is Action.SHOW:
        self._handle_show()
    elif self.orchestrator_config.action is Action.RUN:
        return self._handle_run()