workflows.services.validators

Validation helpers for the workflow wizard payload (server-side).

Attributes

_SEGMENT_CHARS
_ALLOWED_METHODS
_WEBHOOK_MAX_TIMEOUT_SECS

Functions

_error(errors, msg)
_is_dotpath(s)	Return True if s is a valid dot-path like 'serial_number' or 'http.status' (bare path; no prefix).
_is_bare_var_path(s)	Bare variable path (no 'vars.'), at least one segment, dot-separated.
_known_event_triples()	Return set of (handler, protocol_lc, operation) from Events (protocol normalized).
_registered_step_types()	Return the set of registered step type identifiers.
_positive_int(value)
_validate_headers_dict(val)	Accept {str: str|int|float} to match executor (which renders to str).
_validate_webhook_auth(val)	Validate webhook auth object.
_is_valid_from_path(s)	Allow: status | text | json[.a.b.0] | headers[.x_y] (no dashes in dot lookups).
_validate_email_step(idx, params, errors)
_validate_approval_step(idx, params, errors)
_validate_timer_step(idx, params, errors)
_validate_condition_step(idx, params, errors)
_validate_webhook_basic_fields(idx, params, errors)	URL, method, headers, body for webhook step.
_validate_webhook_auth_and_timeout(idx, params, errors)	Auth and timeout for webhook step.
_validate_webhook_result_mapping(idx, params, errors)	result_to/result_source/exports for webhook step.
_validate_webhook_step(idx, params, errors)
_validate_name(payload, errors)
_validate_events(payload, errors)
_validate_single_step(idx, step, registered, errors)	Validate a single step object.
_validate_steps(payload, errors)
_validate_scopes(payload, errors)
validate_wizard_payload(payload)	Validate the wizard JSON (pre-transform).

Module Contents

workflows.services.validators._error(errors, msg)

Parameters:

• errors (list[str])

• msg (str)

Return type:

None

workflows.services.validators._SEGMENT_CHARS

workflows.services.validators._is_dotpath(s)

Return True if s is a valid dot-path like ‘serial_number’ or ‘http.status’ (bare path; no prefix).

Parameters:

s (str)

Return type:

bool

workflows.services.validators._is_bare_var_path(s)

Bare variable path (no ‘vars.’), at least one segment, dot-separated.

Parameters:

s (str)

Return type:

bool

workflows.services.validators._known_event_triples()

Return set of (handler, protocol_lc, operation) from Events (protocol normalized).

Return type:

set[tuple[str, str, str]]

workflows.services.validators._registered_step_types()

Return the set of registered step type identifiers.

Returns:

Step type names as registered with the executor factory.

Return type:

set[str]

workflows.services.validators._positive_int(value)

Parameters:

value (Any)

Return type:

bool

workflows.services.validators._ALLOWED_METHODS

workflows.services.validators._WEBHOOK_MAX_TIMEOUT_SECS = 120

workflows.services.validators._validate_headers_dict(val)

Accept {str: str|int|float} to match executor (which renders to str).

Parameters:

val (Any)

Return type:

bool

workflows.services.validators._validate_webhook_auth(val)

Validate webhook auth object.

Parameters:

val (Any)

Return type:

tuple[bool, str | None]

workflows.services.validators._is_valid_from_path(s)

Allow: status | text | json[.a.b.0] | headers[.x_y] (no dashes in dot lookups).

Parameters:

s (str)

Return type:

bool

workflows.services.validators._validate_email_step(idx, params, errors)

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_approval_step(idx, params, errors)

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_timer_step(idx, params, errors)

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_condition_step(idx, params, errors)

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_webhook_basic_fields(idx, params, errors)

URL, method, headers, body for webhook step.

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_webhook_auth_and_timeout(idx, params, errors)

Auth and timeout for webhook step.

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_webhook_result_mapping(idx, params, errors)

result_to/result_source/exports for webhook step.

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_webhook_step(idx, params, errors)

Parameters:

• idx (int)

• params (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_name(payload, errors)

Parameters:

• payload (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_events(payload, errors)

Parameters:

• payload (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_single_step(idx, step, registered, errors)

Validate a single step object.

Parameters:

• idx (int)

• step (Any)

• registered (set[str])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_steps(payload, errors)

Parameters:

• payload (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators._validate_scopes(payload, errors)

Parameters:

• payload (dict[str, Any])

• errors (list[str])

Return type:

None

workflows.services.validators.validate_wizard_payload(payload)

Validate the wizard JSON (pre-transform).

Parameters:

payload (dict[str, Any]) – Raw wizard configuration payload as a dictionary.

Returns:

A list of human-readable error messages. The list is empty if the payload passes all validation checks.

Return type:

list[str]