yaml_workflow.exceptions

yaml_workflow.exceptions

Custom exceptions for the YAML Workflow Engine.

Classes

ConfigurationError

Bases: WorkflowError

Raised when workflow configuration is invalid or inconsistent.

Source code in src/yaml_workflow/exceptions.py

class ConfigurationError(WorkflowError):
    """Raised when workflow configuration is invalid or inconsistent."""

    pass  # Simple inheritance is often sufficient

FlowError

Bases: WorkflowError

Base exception class for flow-related errors.

Source code in src/yaml_workflow/exceptions.py

class FlowError(WorkflowError):
    """Base exception class for flow-related errors."""

    def __init__(self, message: str):
        super().__init__(f"Flow error: {message}")

FlowNotFoundError

Bases: FlowError

Raised when a specified flow is not found.

Source code in src/yaml_workflow/exceptions.py

class FlowNotFoundError(FlowError):
    """Raised when a specified flow is not found."""

    def __init__(self, flow_name: str):
        super().__init__(f"Flow '{flow_name}' not found")
        self.flow_name = flow_name

FunctionNotFoundError

Bases: StepError

Raised when a function specified in a step cannot be found in the module.

Source code in src/yaml_workflow/exceptions.py

class FunctionNotFoundError(StepError):
    """Raised when a function specified in a step cannot be found in the module."""

    def __init__(self, step_name: str, module_name: str, function_name: str):
        super().__init__(
            step_name, f"Function '{function_name}' not found in module '{module_name}'"
        )

InputResolutionError

Bases: WorkflowRuntimeError

Raised when input variables cannot be resolved.

Source code in src/yaml_workflow/exceptions.py

class InputResolutionError(WorkflowRuntimeError):
    """Raised when input variables cannot be resolved."""

    def __init__(self, step_name: str, variable_name: str, message: str):
        self.step_name = step_name
        self.variable_name = variable_name
        super().__init__(
            f"Failed to resolve input '{variable_name}' in step '{step_name}': {message}"
        )

InputValidationError

Bases: StepError

Raised when step input validation fails.

Source code in src/yaml_workflow/exceptions.py

class InputValidationError(StepError):
    """Raised when step input validation fails."""

    def __init__(self, step_name: str, input_name: str, message: str):
        super().__init__(step_name, f"Invalid input '{input_name}': {message}")

InvalidFlowDefinitionError

Bases: FlowError

Raised when a flow definition is invalid.

Source code in src/yaml_workflow/exceptions.py

class InvalidFlowDefinitionError(FlowError):
    """Raised when a flow definition is invalid."""

    def __init__(self, flow_name: str, reason: str):
        super().__init__(f"Invalid flow '{flow_name}': {reason}")
        self.flow_name = flow_name
        self.reason = reason

ModuleImportError

Bases: WorkflowRuntimeError

Raised when a module cannot be imported.

Source code in src/yaml_workflow/exceptions.py

class ModuleImportError(WorkflowRuntimeError):
    """Raised when a module cannot be imported."""

    pass

ModuleNotFoundError

Bases: StepError

Raised when a module specified in a step cannot be found.

Source code in src/yaml_workflow/exceptions.py

class ModuleNotFoundError(StepError):
    """Raised when a module specified in a step cannot be found."""

    def __init__(self, step_name: str, module_name: str):
        super().__init__(step_name, f"Module '{module_name}' not found")

OutputHandlingError

Bases: WorkflowRuntimeError

Raised when there are issues handling task outputs.

Source code in src/yaml_workflow/exceptions.py

class OutputHandlingError(WorkflowRuntimeError):
    """Raised when there are issues handling task outputs."""

    def __init__(self, step_name: str, message: str):
        self.step_name = step_name
        super().__init__(f"Output handling failed for step '{step_name}': {message}")

OutputValidationError

Bases: StepError

Raised when step output validation fails.

Source code in src/yaml_workflow/exceptions.py

class OutputValidationError(StepError):
    """Raised when step output validation fails."""

    def __init__(self, step_name: str, output_name: str, message: str):
        super().__init__(step_name, f"Invalid output '{output_name}': {message}")

RequiredVariableError

Bases: WorkflowRuntimeError

Raised when a required variable is missing from the context.

Source code in src/yaml_workflow/exceptions.py

class RequiredVariableError(WorkflowRuntimeError):
    """Raised when a required variable is missing from the context."""

    def __init__(self, variable_name: str, step_name: Optional[str] = None):
        self.variable_name = variable_name
        self.step_name = step_name
        location = f" in step '{step_name}'" if step_name else ""
        super().__init__(f"Required variable '{variable_name}' not found{location}")

StepError

Bases: WorkflowError

Base exception class for step-related errors.

Source code in src/yaml_workflow/exceptions.py

class StepError(WorkflowError):
    """Base exception class for step-related errors."""

    def __init__(self, step_name: str, message: str):
        self.step_name = step_name
        super().__init__(f"Error in step '{step_name}': {message}")

StepExecutionError

Bases: StepError

Raised when a step fails during execution.

Source code in src/yaml_workflow/exceptions.py

class StepExecutionError(StepError):
    """Raised when a step fails during execution."""

    def __init__(self, step_name: str, original_error: Exception):
        super().__init__(step_name, f"Execution failed: {str(original_error)}")
        self.original_error = original_error

StepNotInFlowError

Bases: FlowError

Raised when trying to access a step that is not in the current flow.

Source code in src/yaml_workflow/exceptions.py

class StepNotInFlowError(FlowError):
    """Raised when trying to access a step that is not in the current flow."""

    def __init__(self, step_name: str, flow_name: str):
        super().__init__(f"Step '{step_name}' not found in flow '{flow_name}'")
        self.step_name = step_name
        self.flow_name = flow_name

TaskExecutionError

Bases: WorkflowRuntimeError

Raised when a task fails during execution.

Source code in src/yaml_workflow/exceptions.py

class TaskExecutionError(WorkflowRuntimeError):
    """Raised when a task fails during execution."""

    def __init__(
        self,
        step_name: str,
        original_error: Exception,
        task_config: Optional[dict] = None,
    ):
        self.step_name = step_name
        self.original_error = original_error
        self.task_config = task_config
        super().__init__(
            message=f"Task '{step_name}' failed: {str(original_error)}",
            original_error=original_error,
        )

TemplateError

Bases: WorkflowError

Raised when template resolution fails.

Source code in src/yaml_workflow/exceptions.py

class TemplateError(WorkflowError):
    """Raised when template resolution fails."""

    def __init__(self, message: str, original_error: Optional[Exception] = None):
        super().__init__(f"Template error: {message}", original_error=original_error)

VariableNotFoundError

Bases: WorkflowError

Raised when a referenced variable is not found in the workflow context.

Source code in src/yaml_workflow/exceptions.py

class VariableNotFoundError(WorkflowError):
    """Raised when a referenced variable is not found in the workflow context."""

    def __init__(self, variable_name: str):
        super().__init__(f"Variable '{variable_name}' not found in workflow context")

WorkflowDefinitionError

Bases: WorkflowError

Raised when there are issues with the workflow definition YAML.

Source code in src/yaml_workflow/exceptions.py

class WorkflowDefinitionError(WorkflowError):
    """Raised when there are issues with the workflow definition YAML."""

    pass

WorkflowError

Bases: Exception

Base exception class for all workflow-related errors.

Source code in src/yaml_workflow/exceptions.py

class WorkflowError(Exception):
    """Base exception class for all workflow-related errors."""

    def __init__(self, message: str, original_error: Optional[Exception] = None):
        super().__init__(message)
        self.original_error = original_error

WorkflowNotFoundError

Bases: WorkflowError

Raised when a workflow file cannot be found.

Source code in src/yaml_workflow/exceptions.py

class WorkflowNotFoundError(WorkflowError):
    """Raised when a workflow file cannot be found."""

    pass

WorkflowRuntimeError

Bases: WorkflowError

Base class for runtime workflow errors.

Source code in src/yaml_workflow/exceptions.py

class WorkflowRuntimeError(WorkflowError):
    """Base class for runtime workflow errors."""

    pass

WorkflowTimeoutError

Bases: WorkflowError

Raised when a workflow exceeds its timeout limit.

Source code in src/yaml_workflow/exceptions.py

class WorkflowTimeoutError(WorkflowError):
    """Raised when a workflow exceeds its timeout limit."""

    def __init__(self, timeout_seconds: float):
        super().__init__(
            f"Workflow execution exceeded timeout of {timeout_seconds} seconds"
        )

WorkflowValidationError

Bases: WorkflowError

Raised when workflow YAML validation fails.

Source code in src/yaml_workflow/exceptions.py

class WorkflowValidationError(WorkflowError):
    """Raised when workflow YAML validation fails."""

    pass

WorkflowValidationSchema

Schema definitions for workflow validation.

Source code in src/yaml_workflow/exceptions.py

class WorkflowValidationSchema:
    """Schema definitions for workflow validation."""

    REQUIRED_STEP_FIELDS = ["name", "module", "function"]
    OPTIONAL_STEP_FIELDS = [
        "inputs",
        "outputs",
        "condition",
        "error_handling",
        "retry",
        "always_run",
    ]
    VALID_ERROR_HANDLING = ["skip", "fail", "retry", "notify"]