workflows.services.executors.factory

Factory and base types for workflow step executors.

Classes

AbstractStepExecutor	Base class for all step executors.
StepExecutorFactory	Registry-backed factory for step executors.

Module Contents

class workflows.services.executors.factory.AbstractStepExecutor

Base class for all step executors.

execute(instance, signal)

Execute this step.

Parameters:

• instance (workflows.models.WorkflowInstance) – The workflow instance being advanced.

• signal (str | None) – Optional external signal passed to the step.

Returns:

The result of executing the step.

Return type:

workflows.services.types.ExecutorResult

do_execute(instance, signal)

Actual execution hook implemented by concrete executors.

Parameters:

• instance (workflows.models.WorkflowInstance) – The workflow instance being advanced.

• signal (str | None) – Optional external signal passed to the step.

Returns:

The result of executing the step.

Raises:

NotImplementedError – Always; subclasses must implement.

Return type:

workflows.services.types.ExecutorResult

class workflows.services.executors.factory.StepExecutorFactory

Registry-backed factory for step executors.

_registry: ClassVar[dict[str, type[AbstractStepExecutor]]]

classmethod register(step_type, executor_cls)

Register an executor class for a step type.

Parameters:

• step_type (str) – Identifier of the step type.

• executor_cls (type[AbstractStepExecutor]) – Concrete executor class to instantiate for this type.

Return type:

None

classmethod create(step_type)

Create an executor for the given step type.

Parameters:

step_type (str) – Identifier of the step type.

Returns:

An instance of the registered executor class.

Raises:

ValueError – If no executor is registered for step_type.

Return type:

AbstractStepExecutor

classmethod registered_types()

Return the set of registered step type identifiers.

Return type:

set[str]