src/workflows/errors.py

"""
Custom exceptions for workflow validation and execution errors.

This module defines the exception hierarchy for the workflows package, enabling
specific error types to be raised and caught during workflow validation and execution.
Each exception provides detailed error messages to help diagnose and fix issues in
workflow definitions or execution.

Exception hierarchy:
- WorkflowError (base class)
  - UnknownVariableError (missing variable reference)
  - CyclicDependencyError (circular dependencies)
  - FunctionNotFoundError (missing function reference)
"""


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

    This is the parent class for all workflow-specific exceptions and can be used
    to catch any error from the workflows package.
    """

    pass


class UnknownVariableError(WorkflowError):
    """
    Raised when a workflow step references a variable that doesn't exist.

    This typically occurs when a step's input field references a variable that is neither
    provided as an external input nor produced as an output by any previous step.
    """

    def __init__(self, var: str):
        super().__init__(f"Unknown variable referenced: {var}")


class CyclicDependencyError(WorkflowError):
    """
    Raised when a cyclic dependency is detected in a workflow.

    A cyclic dependency occurs when there is a circular reference in the workflow graph,
    such as step A depending on step B, which depends on step A. Such workflows cannot
    be executed because there's no valid order to process the steps.
    """

    def __init__(self):
        super().__init__("Cyclic dependency detected in workflow")


class FunctionNotFoundError(WorkflowError):
    """
    Raised when a referenced function cannot be found during workflow execution.

    This typically occurs when a step references a function that doesn't exist in
    the available function registry or namespace.
    """

    def __init__(self, func_name: str):
        super().__init__(f"Function not found: {func_name}")