workflows.services.executors

Built-in workflow step executors registry.

Submodules

• workflows.services.executors.approval
• workflows.services.executors.email
• workflows.services.executors.factory
• workflows.services.executors.webhook

Classes

ApprovalExecutor	Approval step executor.
EmailExecutor	Send an email using either a named template or a simple subject/body.
StepExecutorFactory	Registry-backed factory for step executors.
WebhookExecutor	Execute an outbound HTTP call and optionally export values to $vars.

Package Contents

class workflows.services.executors.ApprovalExecutor

Bases: workflows.services.executors.factory.AbstractStepExecutor

Approval step executor.

First encounter (no signal) → AWAITING (AwaitingApproval).

• On “reject” → REJECTED (terminal).

• On “approve”:

  • If this is the last Approval step → APPROVED.

  • Otherwise → PASSED.

do_execute(instance, signal)

Execute the approval step and return the resulting state.

Parameters:

• instance (workflows.models.WorkflowInstance) – Workflow instance being processed.

• signal (str | None) – External decision signal such as approve or reject.

Returns:

ExecutorResult describing the new workflow state and step context.

Return type:

workflows.services.types.ExecutorResult

class workflows.services.executors.EmailExecutor

Bases: workflows.services.executors.factory.AbstractStepExecutor

Send an email using either a named template or a simple subject/body.

do_execute(instance, _signal)

Execute the email step.

Parameters:

• instance (workflows.models.WorkflowInstance) – Workflow instance being executed.

• _signal (str | None) – Optional signal (unused for email steps).

Returns:

ExecutorResult describing the outcome of sending the email.

Return type:

workflows.services.types.ExecutorResult

static _prepare_recipients_and_context(params, instance)

Prepare normalized recipients and template context.

Parameters:

• params (dict[str, Any])

• instance (workflows.models.WorkflowInstance)

Return type:

EmailParts

static _error_result(message, *, outputs=None)

Return a standardized FAILED result for this email step.

Parameters:

• message (str) – Human-readable error description.

• outputs (dict[str, Any] | None) – Optional additional output fields to include in the context.

Returns:

ExecutorResult with FAILED state and error details.

Return type:

workflows.services.types.ExecutorResult

_send_template_email(params, template_key, parts)

Handle template-based email sending.

Parameters:

• params (dict[str, Any])

• template_key (str)

• parts (EmailParts)

Return type:

workflows.services.types.ExecutorResult

_send_custom_email(params, parts)

Handle plain-text email sending.

Parameters:

• params (dict[str, Any])

• parts (EmailParts)

Return type:

workflows.services.types.ExecutorResult

class workflows.services.executors.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]

class workflows.services.executors.WebhookExecutor

Bases: workflows.services.executors.factory.AbstractStepExecutor

Execute an outbound HTTP call and optionally export values to $vars.

• URL, headers, and string body templating with Django templates using a ‘ctx’ dict

• Supports method, headers, body, auth (basic|bearer), timeoutSecs

• result_to/result_source for whole-response capture

• fine-grained exports: [{“from_path”:”json.foo”,”to_path”:”serial”}] # note: bare key allowed

• Stores per-step context and returns a flat vars map for $vars merging.

to_path:

Accepts either “serial” or “vars.serial” (we strip optional “vars.” and store under $vars).

do_execute(instance, _signal)

Execute the webhook step.

Parameters:

• instance (workflows.models.WorkflowInstance) – Workflow instance being executed.

• _signal (str | None) – Optional signal (unused for webhook steps).

Returns:

ExecutorResult describing step outcome and exported vars.

Return type:

workflows.services.types.ExecutorResult