Skip to main content

Raising Custom Exceptions

When a task in flyte-sdk encounters an unhandled exception, the system automatically wraps it in a CustomError to ensure consistent error reporting and categorization. You can also explicitly use CustomError to provide structured error codes and messages for specific failure scenarios in your tasks.

Wrapping Standard Exceptions​

The flyte-sdk runtime automatically catches standard Python exceptions and converts them into CustomError instances. This is implemented in _internal/runtime/taskrunner.py within the run_task function:

from errors import CustomError, RuntimeSystemError, RuntimeUnknownError, RuntimeUserError

# Internal logic in _internal/runtime/taskrunner.py
try:
    outputs = await task.execute(**inputs)
    return outputs, None
except RuntimeSystemError as e:
    return {}, e
except RuntimeUnknownError as e:
    return {}, e
except RuntimeUserError as e:
    return {}, e
except Exception as e:
    # Any other exception is wrapped as a CustomError
    return {}, CustomError.from_exception(e)

When CustomError.from_exception(e) is called, it uses the exception's class name as the error code and the exception's string representation as the message.

Defining and Raising Custom Errors​

You can raise a CustomError directly in your task code to provide a specific error code that can be used for downstream monitoring or conditional logic.

from errors import CustomError

def my_task(val: int):
    if val < 0:
        # Explicitly raise a CustomError with a specific code
        raise CustomError(
            code="INVALID_INPUT_VALUE",
            message=f"Input value must be non-negative, received: {val}"
        )
    return val * 2

Using the from_exception Helper​

If you catch an exception and want to re-raise it as a CustomError manually (for example, to ensure it is categorized as a "user" error type), use the from_exception class method.

from errors import CustomError

def process_data(data: dict):
    try:
        return data["required_key"]
    except KeyError as e:
        # Converts KeyError to a CustomError with code "KeyError"
        raise CustomError.from_exception(e)

Error Structure and Categorization​

The CustomError class inherits from RuntimeUserError, which means it is always categorized with the user error kind. This helps flyte-sdk distinguish between failures caused by the user's code/logic versus infrastructure or system-level issues.

The constructor for CustomError in errors.py is defined as:

class CustomError(RuntimeUserError):
    def __init__(self, code: str, message: str):
        super().__init__(code, message, "user")
  • code: A string identifier for the error (e.g., "ValidationError", "DatabaseConnectionError").
  • message: A human-readable description of what went wrong.
  • kind: Automatically set to "user".

Troubleshooting: Uncaught Exceptions​

If your task fails with a CustomError but you didn't explicitly raise one, check the error code in the logs. Because flyte-sdk wraps all unknown exceptions, the code will match the name of the Python exception that was originally raised (e.g., ValueError, TypeError, or RuntimeError). This wrapping occurs at the boundary of the task execution to ensure the runtime can always process the error object consistently.