Skip to main content

Raising Custom Errors in Tasks

Raise a business error from a task​

To propagate a business-rule failure as a Flyte user error, raise CustomError with an application-specific code and message:

from flyte.errors import CustomError


def require_positive(value: int) -> int:
    if value <= 0:
        raise CustomError("NonPositiveValue", "value must be greater than zero")
    return value

The task example above is instructional: flyte-sdk does not contain a checked-in user task example that raises CustomError. It uses the constructor defined in errors.py:

class CustomError(RuntimeUserError):
    def __init__(self, code: str, message: str):
        super().__init__(code, message, "user")

CustomError takes exactly code and message. It subclasses RuntimeUserError, whose general constructor is RuntimeUserError(code, message, worker=None). RuntimeUserError passes the fixed kind "user" to BaseRuntimeError, so both classes are normal Python RuntimeError exceptions with structured code, kind, and worker attributes.

Inspect the structured fields​

BaseRuntimeError initializes the Python exception with the root-cause message and stores the other fields separately. Consequently, str(error) is the message; it does not include the code or worker automatically:

from flyte.errors import CustomError

error = CustomError("InventoryUnavailable", "inventory service returned no stock")

assert error.code == "InventoryUnavailable"
assert error.kind == "user"
assert str(error) == "inventory service returned no stock"

In the current errors.py implementation, CustomError calls RuntimeUserError with the positional third argument "user". Since that argument is RuntimeUserError's worker parameter, error.worker is also set to "user" for a CustomError instance. CustomError does not expose a separate worker argument. If you need to construct a user error with an explicitly chosen worker, use RuntimeUserError directly:

from flyte.errors import RuntimeUserError

error = RuntimeUserError(
    "WorkerValidationFailed",
    "the worker rejected the input",
    worker="validation-worker",
)

Convert an ordinary exception to CustomError​

Use CustomError.from_exception when a task boundary needs to translate an ordinary Python exception into the structured custom-error type:

from flyte.errors import CustomError

try:
    raise ValueError("account is not active")
except Exception as exc:
    error = CustomError.from_exception(exc)

assert error.code == "ValueError"
assert str(error) == "account is not active"
assert error.kind == "user"

The class method uses e.__class__.__name__ verbatim for code and str(e) for message:

@classmethod
def from_exception(cls, e: Exception):
    return cls(e.__class__.__name__, str(e))

This means the resulting code follows the Python exception class name rather than an application-defined stable code. Construct CustomError directly when consumers need a stable business-error code.

How task execution preserves the error​

TaskTemplate.__call__ in _task.py re-raises existing RuntimeSystemError and RuntimeUserError instances. An explicitly raised CustomError is therefore preserved because it is a RuntimeUserError. Other exceptions at that boundary become RuntimeUserError(type(e).__name__, str(e)) and are chained from the original exception:

try:
    # TaskTemplate.__call__ invokes the task/controller path here.
    result = task(*args, **kwargs)
except RuntimeUserError:
    raise

The relevant exception policy in TaskTemplate.__call__ is:

try:
    result = task(*args, **kwargs)
except RuntimeSystemError:
    raise
except RuntimeUserError:
    raise
except Exception as e:
    raise RuntimeUserError(type(e).__name__, str(e)) from e

For the V2 runtime path, _internal/runtime/taskrunner.py defines run_task. It returns a pair containing task outputs and an optional exception. Known runtime errors—including CustomError—are returned unchanged; an otherwise unhandled exception is converted with CustomError.from_exception:

def run_task(task, inputs):
    try:
        outputs = 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:
        return {}, CustomError.from_exception(e)

The production function is asynchronous and has the signature run_task(tctx: TaskContext, controller: Controller, task: TaskTemplate, inputs: Dict[str, Any]); the shortened wrapper above is only intended to make the exception policy easy to read. Do not raise CustomError for an infrastructure failure. Its hierarchy is the user-error hierarchy and its serialized kind is USER; RuntimeSystemError is the separate class used for system failures.

Cross the execution-error protocol boundary​

When a runtime error is converted for the execution service, _internal/runtime/convert.py checks RuntimeUserError and creates an execution_pb2.ExecutionError with USER kind, the error's code, str(error) as message, and the error's worker:

def convert_from_native_to_error(err: BaseException) -> Error:
    if isinstance(err, flyte.errors.RuntimeUnknownError):
        return Error(
            err=execution_pb2.ExecutionError(
                kind=execution_pb2.ExecutionError.UNKNOWN,
                code=err.code,
                message=str(err),
                worker=err.worker,
            )
        )
    elif isinstance(err, flyte.errors.RuntimeUserError):
        return Error(
            err=execution_pb2.ExecutionError(
                kind=execution_pb2.ExecutionError.USER,
                code=err.code,
                message=str(err),
                worker=err.worker,
            )
        )

CustomError matches the RuntimeUserError branch. The structured values—not a formatted version of the exception text—are what the converter places in the protocol error. On the way back, convert_error_to_native reconstructs a native exception for a USER error. Recognized code markers produce specialized user-error subclasses; an unrecognized code produces RuntimeUserError with the protocol code, message, and worker:

def convert_user_error(err):
    match err.kind:
        case execution_pb2.ExecutionError.USER:
            if "OOM" in err.code.upper():
                return flyte.errors.OOMError(code=user_code, message=err.message, worker=err.worker)
            return flyte.errors.RuntimeUserError(code=user_code, message=err.message, worker=err.worker)

The production convert_error_to_native also strips a server-injected server_code|user_code prefix before constructing the native error. A returned remote error is generally reconstructed as RuntimeUserError, not as CustomError; the protocol preserves the USER classification and fields, but the conversion fallback does not identify arbitrary custom codes as the CustomError subclass.

Troubleshoot custom error codes​

  • Use meaningful, stable codes. CustomError.from_exception derives the code from the Python class name and can produce an empty or weak message when str(e) is unhelpful. Direct construction gives the business failure a deliberate code.
  • Avoid reserved substrings in custom codes. convert_error_to_native classifies USER errors using substring checks such as "OOM" in err.code.upper() and checks for markers including Interrupted, RetriesExhausted, InvalidImageName, and ImagePullBackOff. A custom code containing one of those markers can be reconstructed as a specialized error instead of the generic user-error fallback.
  • Do not pass worker to CustomError. Its signature is only (code, message). In the current source, its positional delegation sets worker to "user"; use RuntimeUserError(..., worker=...) when a specific worker value is required.
  • Keep system failures in the system hierarchy. convert_from_native_to_error serializes RuntimeUserError subclasses as USER errors and RuntimeSystemError as SYSTEM errors. Choose the class that matches the failure you are raising.