flytekit.exceptions.utils
| Property |
Type |
Description |
SOURCE_CODE |
str |
|
def annotate_exception_with_code(
exception: flytekit.exceptions.user.FlyteUserException,
fn: typing.Callable,
param_name: typing.Optional[str] = None,
) -> flytekit.exceptions.user.FlyteUserException
Annotate the exception with the source code, and will be printed in the rich panel.
| Parameter |
Type |
Description |
exception |
flytekit.exceptions.user.FlyteUserException |
The exception to be annotated. |
fn |
typing.Callable |
The function where the parameter is defined. |
param_name |
typing.Optional[str] |
The name of the parameter in the function signature. For example: exception: TypeError, ‘a’ has no type. Please add a type annotation to the input parameter. param_name: a, the position that arrow will point to. fn: <function wf at 0x1065227a0> ╭─ TypeError ────────────────────────────────────────────────────────────────────────────────────╮ │ 23 @workflow(on_failure=t2) │ │ │ 24 def wf(b: int = 3, a=4): │ │ # ^ ‘a’ has no type. Please add a type annotation to the input parameter. │ ╰────────────────────────────────────────────────────────────────────────────────────────────────╯ |
def get_source_code_from_fn(
fn: typing.Callable,
param_name: typing.Optional[str] = None,
) -> (<class 'str'>, <class 'int'>)
Get the source code of the function and the column offset of the parameter defined in the input signature.
| Parameter |
Type |
Description |
fn |
typing.Callable |
|
param_name |
typing.Optional[str] |
|
