Skip to content

Hooks

zenml.hooks

(Lifecycle) Hooks for steps and pipelines.

Attributes

__all__ = ['alerter_success_hook', 'alerter_failure_hook', 'run_hook'] module-attribute

Functions:

alerter_failure_hook(exception: BaseException) -> None

Standard failure hook that executes after step fails.

This hook uses any BaseAlerter that is configured within the active stack to post a message.

Parameters:

Name Type Description Default
exception BaseException

Original exception that lead to step failing.

 required
Source code in src/zenml/hooks/alerter_hooks.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def alerter_failure_hook(exception: BaseException) -> None:
    """Standard failure hook that executes after step fails.

    This hook uses any `BaseAlerter` that is configured within the active stack to post a message.

    Args:
        exception: Original exception that lead to step failing.
    """
    context = get_step_context()
    alerter = Client().active_stack.alerter
    if alerter:
        output_captured = io.StringIO()
        original_stdout = sys.stdout
        sys.stdout = output_captured
        console = Console()
        console.print_exception(show_locals=False)

        sys.stdout = original_stdout
        rich_traceback = output_captured.getvalue()

        message = "*Failure Hook Notification! Step failed!*" + "\n\n"
        message += f"Pipeline name: `{context.pipeline.name}`" + "\n"
        message += f"Run name: `{context.pipeline_run.name}`" + "\n"
        message += f"Step name: `{context.step_run.name}`" + "\n"
        message += f"Parameters: `{context.step_run.config.parameters}`" + "\n"
        message += (
            f"Exception: `({type(exception)}) {rich_traceback}`" + "\n\n"
        )
        alerter.post(message)
    else:
        logger.warning(
            "Specified standard failure hook but no alerter configured in the stack. Skipping.."
        )

alerter_success_hook() -> None

Standard success hook that executes after step finishes successfully.

This hook uses any BaseAlerter that is configured within the active stack to post a message.

Source code in src/zenml/hooks/alerter_hooks.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def alerter_success_hook() -> None:
    """Standard success hook that executes after step finishes successfully.

    This hook uses any `BaseAlerter` that is configured within the active stack to post a message.
    """
    context = get_step_context()
    alerter = Client().active_stack.alerter
    if alerter:
        message = (
            "*Success Hook Notification! Step completed successfully*" + "\n\n"
        )
        message += f"Pipeline name: `{context.pipeline.name}`" + "\n"
        message += f"Run name: `{context.pipeline_run.name}`" + "\n"
        message += f"Step name: `{context.step_run.name}`" + "\n"
        message += f"Parameters: `{context.step_run.config.parameters}`" + "\n"
        alerter.post(message)
    else:
        logger.warning(
            "Specified standard success hook but no alerter configured in the stack. Skipping.."
        )

run_hook(func: Union[Callable[..., Any], Source], *args: Any, kwargs: Optional[Dict[str, Any]] = None, name: Optional[str] = None, hook_type: HookType = HookType.CUSTOM, store_return: bool = False, track: bool = True) -> Any

Run a hook, optionally recording the invocation.

Parameters:

Name Type Description Default
func Union[Callable[..., Any], Source]

The hook function or a source to load it from.

 required
*args Any

Positional arguments for the hook function.

 ()
kwargs Optional[Dict[str, Any]]

Keyword arguments for the hook function.

 None
name Optional[str]

Custom event name for the invocation.

 None
hook_type HookType

Type of the hook invocation.

 CUSTOM
store_return bool

Whether to materialize the return value as outputs.

 False
track bool

Whether to record the invocation.

 True

Raises:

Type Description
BaseException

Any exception raised by the hook function.

Returns:

Type Description
Any

The return value of the hook function.
Source code in src/zenml/hooks/execution.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def run_hook(
    func: Union[Callable[..., Any], Source],
    *args: Any,
    kwargs: Optional[Dict[str, Any]] = None,
    name: Optional[str] = None,
    hook_type: HookType = HookType.CUSTOM,
    store_return: bool = False,
    track: bool = True,
) -> Any:
    """Run a hook, optionally recording the invocation.

    Args:
        func: The hook function or a source to load it from.
        *args: Positional arguments for the hook function.
        kwargs: Keyword arguments for the hook function.
        name: Custom event name for the invocation.
        hook_type: Type of the hook invocation.
        store_return: Whether to materialize the return value as outputs.
        track: Whether to record the invocation.

    Raises:
        BaseException: Any exception raised by the hook function.

    Returns:
        The return value of the hook function.
    """
    if isinstance(func, Source):
        loaded_func = source_utils.load(func)
        import_path: Optional[str] = func.import_path
    else:
        loaded_func = func
        try:
            import_path = source_utils.resolve(func).import_path
        except Exception:
            import_path = None

    _validate_hook_name(name)
    resolved_name = name or _default_hook_name(loaded_func, hook_type)

    start_time = utc_now()
    status = ExecutionStatus.COMPLETED
    outputs: Optional[Dict[str, Any]] = None
    exception_info: Optional[ExceptionInfo] = None

    logs_context = setup_hook_logging_context(hook_type, resolved_name)
    logs_id = logs_context.log_model.id if logs_context is not None else None

    with logs_context or nullcontext():
        try:
            try:
                if async_utils.is_async_callable(loaded_func):
                    result = async_utils.run_coroutine_isolated(
                        loaded_func(*args, **(kwargs or {}))
                    )
                else:
                    result = loaded_func(*args, **(kwargs or {}))
            except BaseException as e:
                status = ExecutionStatus.FAILED
                exception_info = collect_exception_information(
                    e, user_func=loaded_func
                )
                raise
            # Output parsing runs after the hook succeeded, so a parsing error
            # does not mark the invocation as FAILED or discard the hook's
            # status.
            if store_return:
                outputs = _parse_hook_outputs(loaded_func, result)
            return result
        finally:
            if track:
                try:
                    record_hook_invocation(
                        name=resolved_name,
                        hook_type=hook_type,
                        source=import_path,
                        outputs=outputs,
                        start_time=start_time,
                        end_time=utc_now(),
                        status=status,
                        exception_info=exception_info,
                        logs_id=logs_id,
                    )
                except Exception as e:
                    logger.error(
                        "Failed to record `%s` hook invocation: %s",
                        hook_type.value,
                        e,
                    )

Modules

alerter_hooks

Functionality for standard hooks.

Classes
Functions:
alerter_failure_hook(exception: BaseException) -> None

Standard failure hook that executes after step fails.

This hook uses any BaseAlerter that is configured within the active stack to post a message.

Parameters:

Name Type Description Default
exception BaseException

Original exception that lead to step failing.

 required
Source code in src/zenml/hooks/alerter_hooks.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def alerter_failure_hook(exception: BaseException) -> None:
    """Standard failure hook that executes after step fails.

    This hook uses any `BaseAlerter` that is configured within the active stack to post a message.

    Args:
        exception: Original exception that lead to step failing.
    """
    context = get_step_context()
    alerter = Client().active_stack.alerter
    if alerter:
        output_captured = io.StringIO()
        original_stdout = sys.stdout
        sys.stdout = output_captured
        console = Console()
        console.print_exception(show_locals=False)

        sys.stdout = original_stdout
        rich_traceback = output_captured.getvalue()

        message = "*Failure Hook Notification! Step failed!*" + "\n\n"
        message += f"Pipeline name: `{context.pipeline.name}`" + "\n"
        message += f"Run name: `{context.pipeline_run.name}`" + "\n"
        message += f"Step name: `{context.step_run.name}`" + "\n"
        message += f"Parameters: `{context.step_run.config.parameters}`" + "\n"
        message += (
            f"Exception: `({type(exception)}) {rich_traceback}`" + "\n\n"
        )
        alerter.post(message)
    else:
        logger.warning(
            "Specified standard failure hook but no alerter configured in the stack. Skipping.."
        )
alerter_success_hook() -> None

Standard success hook that executes after step finishes successfully.

This hook uses any BaseAlerter that is configured within the active stack to post a message.

Source code in src/zenml/hooks/alerter_hooks.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def alerter_success_hook() -> None:
    """Standard success hook that executes after step finishes successfully.

    This hook uses any `BaseAlerter` that is configured within the active stack to post a message.
    """
    context = get_step_context()
    alerter = Client().active_stack.alerter
    if alerter:
        message = (
            "*Success Hook Notification! Step completed successfully*" + "\n\n"
        )
        message += f"Pipeline name: `{context.pipeline.name}`" + "\n"
        message += f"Run name: `{context.pipeline_run.name}`" + "\n"
        message += f"Step name: `{context.step_run.name}`" + "\n"
        message += f"Parameters: `{context.step_run.config.parameters}`" + "\n"
        alerter.post(message)
    else:
        logger.warning(
            "Specified standard success hook but no alerter configured in the stack. Skipping.."
        )

execution

Hook execution.

Classes
Functions:
run_hook(func: Union[Callable[..., Any], Source], *args: Any, kwargs: Optional[Dict[str, Any]] = None, name: Optional[str] = None, hook_type: HookType = HookType.CUSTOM, store_return: bool = False, track: bool = True) -> Any

Run a hook, optionally recording the invocation.

Parameters:

Name Type Description Default
func Union[Callable[..., Any], Source]

The hook function or a source to load it from.

 required
*args Any

Positional arguments for the hook function.

 ()
kwargs Optional[Dict[str, Any]]

Keyword arguments for the hook function.

 None
name Optional[str]

Custom event name for the invocation.

 None
hook_type HookType

Type of the hook invocation.

 CUSTOM
store_return bool

Whether to materialize the return value as outputs.

 False
track bool

Whether to record the invocation.

 True

Raises:

Type Description
BaseException

Any exception raised by the hook function.

Returns:

Type Description
Any

The return value of the hook function.
Source code in src/zenml/hooks/execution.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def run_hook(
    func: Union[Callable[..., Any], Source],
    *args: Any,
    kwargs: Optional[Dict[str, Any]] = None,
    name: Optional[str] = None,
    hook_type: HookType = HookType.CUSTOM,
    store_return: bool = False,
    track: bool = True,
) -> Any:
    """Run a hook, optionally recording the invocation.

    Args:
        func: The hook function or a source to load it from.
        *args: Positional arguments for the hook function.
        kwargs: Keyword arguments for the hook function.
        name: Custom event name for the invocation.
        hook_type: Type of the hook invocation.
        store_return: Whether to materialize the return value as outputs.
        track: Whether to record the invocation.

    Raises:
        BaseException: Any exception raised by the hook function.

    Returns:
        The return value of the hook function.
    """
    if isinstance(func, Source):
        loaded_func = source_utils.load(func)
        import_path: Optional[str] = func.import_path
    else:
        loaded_func = func
        try:
            import_path = source_utils.resolve(func).import_path
        except Exception:
            import_path = None

    _validate_hook_name(name)
    resolved_name = name or _default_hook_name(loaded_func, hook_type)

    start_time = utc_now()
    status = ExecutionStatus.COMPLETED
    outputs: Optional[Dict[str, Any]] = None
    exception_info: Optional[ExceptionInfo] = None

    logs_context = setup_hook_logging_context(hook_type, resolved_name)
    logs_id = logs_context.log_model.id if logs_context is not None else None

    with logs_context or nullcontext():
        try:
            try:
                if async_utils.is_async_callable(loaded_func):
                    result = async_utils.run_coroutine_isolated(
                        loaded_func(*args, **(kwargs or {}))
                    )
                else:
                    result = loaded_func(*args, **(kwargs or {}))
            except BaseException as e:
                status = ExecutionStatus.FAILED
                exception_info = collect_exception_information(
                    e, user_func=loaded_func
                )
                raise
            # Output parsing runs after the hook succeeded, so a parsing error
            # does not mark the invocation as FAILED or discard the hook's
            # status.
            if store_return:
                outputs = _parse_hook_outputs(loaded_func, result)
            return result
        finally:
            if track:
                try:
                    record_hook_invocation(
                        name=resolved_name,
                        hook_type=hook_type,
                        source=import_path,
                        outputs=outputs,
                        start_time=start_time,
                        end_time=utc_now(),
                        status=status,
                        exception_info=exception_info,
                        logs_id=logs_id,
                    )
                except Exception as e:
                    logger.error(
                        "Failed to record `%s` hook invocation: %s",
                        hook_type.value,
                        e,
                    )
run_lifecycle_hook(hook_source: Optional[Source], hook_type: HookType, optional_args: Optional[Tuple[Any, ...]] = None) -> None

Run a configured lifecycle hook, recording it and swallowing errors.

Parameters:

Name Type Description Default
hook_source Optional[Source]

The source of the hook function, or None to do nothing.

 required
hook_type HookType

The type of the lifecycle hook.

 required
optional_args Optional[Tuple[Any, ...]]

Candidate arguments offered to the hook in order. As many leading arguments as the hook signature accepts are bound.

 None
Source code in src/zenml/hooks/execution.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def run_lifecycle_hook(
    hook_source: Optional[Source],
    hook_type: HookType,
    optional_args: Optional[Tuple[Any, ...]] = None,
) -> None:
    """Run a configured lifecycle hook, recording it and swallowing errors.

    Args:
        hook_source: The source of the hook function, or None to do nothing.
        hook_type: The type of the lifecycle hook.
        optional_args: Candidate arguments offered to the hook in order. As
            many leading arguments as the hook signature accepts are bound.
    """
    if hook_source is None:
        return

    # The signature was already validated when the hook was configured, so we
    # only bind the leading values it accepts and run it.
    try:
        func = source_utils.load(hook_source)
        bound_args = bind_optional_hook_args(func, optional_args or ())
        store_return = handle_bool_env_var(
            ENV_ZENML_TRACK_LIFECYCLE_HOOK_OUTPUTS, default=False
        )
        run_hook(
            func, *bound_args, hook_type=hook_type, store_return=store_return
        )
    except Exception as e:
        logger.error(
            "Failed to run `%s` hook from source `%s`: %s",
            hook_type.value,
            hook_source,
            e,
        )
setup_hook_logging_context(hook_type: HookType, name: Optional[str]) -> Optional[LoggingContext]

Setup a logging context that captures a hook's output.

Parameters:

Name Type Description Default
hook_type HookType

The type of the hook invocation.

 required
name Optional[str]

The resolved name of the hook invocation.

 required

Returns:

Type Description
Optional[LoggingContext]

The logging context.
Source code in src/zenml/hooks/execution.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def setup_hook_logging_context(
    hook_type: HookType,
    name: Optional[str],
) -> Optional["LoggingContext"]:
    """Setup a logging context that captures a hook's output.

    Args:
        hook_type: The type of the hook invocation.
        name: The resolved name of the hook invocation.

    Returns:
        The logging context.
    """
    from zenml.execution.pipeline.dynamic.run_context import (
        DynamicPipelineRunContext,
    )
    from zenml.steps.step_context import StepContext
    from zenml.utils.logging_utils import (
        is_pipeline_logging_enabled,
        is_step_logging_enabled,
        setup_logging_context,
    )

    try:
        source = _get_log_source_name(hook_type, name)
        step_context = StepContext.get()
        if step_context is not None:
            if is_step_logging_enabled(
                step_configuration=step_context.step_run.config,
                pipeline_configuration=step_context.pipeline_run.config,
            ):
                return setup_logging_context(
                    source=source,
                    step_run=step_context.step_run,
                    pipeline_run=step_context.pipeline_run,
                    block_on_exit=False,
                )
        else:
            run_context = DynamicPipelineRunContext.get()
            if run_context is not None and is_pipeline_logging_enabled(
                pipeline_configuration=run_context.run.config,
            ):
                return setup_logging_context(
                    source=source,
                    pipeline_run=run_context.run,
                    block_on_exit=False,
                )
    except Exception as e:
        logger.debug("Failed to set up hook logging context: %s", e)
    return None
Modules

recording

Hook invocation recording.

Classes
Functions:
record_hook_invocation(name: Optional[str] = None, hook_type: HookType = HookType.CUSTOM, source: Optional[str] = None, outputs: Optional[Dict[str, Any]] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, status: ExecutionStatus = ExecutionStatus.COMPLETED, exception_info: Optional[ExceptionInfo] = None, logs_id: Optional[UUID] = None) -> HookInvocationResponse

Record a hook invocation for an already-completed event.

Parameters:

Name Type Description Default
name Optional[str]

Custom event name for the invocation.

 None
hook_type HookType

Type of the hook invocation.

 CUSTOM
source Optional[str]

Resolved source of the hook function.

 None
outputs Optional[Dict[str, Any]]

Output values to materialize as artifacts.

 None
start_time Optional[datetime]

Start time of the event.

 None
end_time Optional[datetime]

End time of the event.

 None
status ExecutionStatus

Terminal status of the event.

 COMPLETED
exception_info Optional[ExceptionInfo]

Exception information when the event failed.

 None
logs_id Optional[UUID]

ID of the logs entry to link to the invocation.

 None

Returns:

Type Description
HookInvocationResponse

The recorded hook invocation.
Source code in src/zenml/hooks/recording.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def record_hook_invocation(
    name: Optional[str] = None,
    hook_type: HookType = HookType.CUSTOM,
    source: Optional[str] = None,
    outputs: Optional[Dict[str, Any]] = None,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
    status: ExecutionStatus = ExecutionStatus.COMPLETED,
    exception_info: Optional[ExceptionInfo] = None,
    logs_id: Optional[UUID] = None,
) -> HookInvocationResponse:
    """Record a hook invocation for an already-completed event.

    Args:
        name: Custom event name for the invocation.
        hook_type: Type of the hook invocation.
        source: Resolved source of the hook function.
        outputs: Output values to materialize as artifacts.
        start_time: Start time of the event.
        end_time: End time of the event.
        status: Terminal status of the event.
        exception_info: Exception information when the event failed.
        logs_id: ID of the logs entry to link to the invocation.

    Returns:
        The recorded hook invocation.
    """
    pipeline_run_id, step_run_id = _resolve_hook_context()

    hook_invocation_id = uuid4()
    now = utc_now()

    materialized_outputs = _materialize_hook_outputs(
        outputs=outputs or {},
        pipeline_run_id=pipeline_run_id,
        hook_invocation_id=hook_invocation_id,
    )

    client = Client()
    request = HookInvocationRequest(
        id=hook_invocation_id,
        project=client.active_project.id,
        hook_type=hook_type,
        name=name,
        status=status,
        start_time=start_time or now,
        end_time=end_time or now,
        source=source,
        pipeline_run_id=pipeline_run_id,
        step_run_id=step_run_id,
        outputs=materialized_outputs,
        exception_info=exception_info,
        logs_id=logs_id,
    )
    return client.zen_store.create_hook_invocation(request)
Modules

validation

Hook resolution and argument validation.

Classes
Functions:
bind_optional_hook_args(func: Callable[..., Any], candidate_args: Tuple[Any, ...]) -> Tuple[Any, ...]

Binds as many leading candidate arguments as the signature accepts.

Parameters:

Name Type Description Default
func Callable[..., Any]

The hook function.

 required
candidate_args Tuple[Any, ...]

Candidate arguments in order.

 required

Returns:

Type Description
Tuple[Any, ...]

The leading candidate arguments the signature can accept.
Source code in src/zenml/hooks/validation.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def bind_optional_hook_args(
    func: Callable[..., Any], candidate_args: Tuple[Any, ...]
) -> Tuple[Any, ...]:
    """Binds as many leading candidate arguments as the signature accepts.

    Args:
        func: The hook function.
        candidate_args: Candidate arguments in order.

    Returns:
        The leading candidate arguments the signature can accept.
    """
    from zenml.steps.utils import get_resolved_signature

    signature = get_resolved_signature(func)
    positional = 0
    for parameter in signature.parameters.values():
        if parameter.kind is inspect.Parameter.VAR_POSITIONAL:
            return candidate_args

        if parameter.kind in (
            inspect.Parameter.POSITIONAL_ONLY,
            inspect.Parameter.POSITIONAL_OR_KEYWORD,
        ):
            positional += 1

    return candidate_args[:positional]
resolve_and_validate_hook(hook: Union[HookSpecification, InitHookSpecification], *args: Any, kwargs: Optional[Dict[str, Any]] = None) -> Tuple[Source, Dict[str, Any]]

Resolve a hook and validate it can be called with the given arguments.

Parameters:

Name Type Description Default
hook Union[HookSpecification, InitHookSpecification]

Hook function or source.

 required
*args Any

Positional arguments offered to the hook. Only the leading arguments the signature accepts are bound and validated.

 ()
kwargs Optional[Dict[str, Any]]

Keyword arguments passed to the hook.

 None

Raises:

Type Description
ValueError

If hook is not a valid callable.
HookValidationException

If the hook cannot accept the arguments.

Returns:

Type Description
Tuple[Source, Dict[str, Any]]

The hook source and the validated, JSON-safe keyword arguments.
Source code in src/zenml/hooks/validation.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def resolve_and_validate_hook(
    hook: Union["HookSpecification", "InitHookSpecification"],
    *args: Any,
    kwargs: Optional[Dict[str, Any]] = None,
) -> Tuple[Source, Dict[str, Any]]:
    """Resolve a hook and validate it can be called with the given arguments.

    Args:
        hook: Hook function or source.
        *args: Positional arguments offered to the hook. Only the leading
            arguments the signature accepts are bound and validated.
        kwargs: Keyword arguments passed to the hook.

    Raises:
        ValueError: If `hook` is not a valid callable.
        HookValidationException: If the hook cannot accept the arguments.

    Returns:
        The hook source and the validated, JSON-safe keyword arguments.
    """
    func = source_utils.load(hook) if isinstance(hook, (str, Source)) else hook
    if not callable(func):
        raise ValueError(f"{func} is not a valid function.")

    bound_args = bind_optional_hook_args(func, args)
    # Allow arbitrary types if an exception arg is provided. If validating an
    # init hook which allows user-provided arguments, we only allow
    # JSON-serializable arguments.
    config = ConfigDict(arbitrary_types_allowed=bool(bound_args))
    try:
        validated_kwargs = validate_function_args(
            func, config, *bound_args, **(kwargs or {})
        )
    except (ValidationError, TypeError) as e:
        raise HookValidationException(
            f"Failed to validate hook arguments for {func}: {e}\n"
            "Please observe the following guidelines:\n"
            "- the success hook takes no arguments\n"
            "- the failure hook optionally takes a single `BaseException`"
            " argument\n"
            "- the on_start hook takes no arguments\n"
            "- the on_end hook optionally takes a single `BaseException`"
            " argument\n"
            "- the on_pause hook takes no arguments\n"
            "- the on_resume hook takes no arguments\n"
            "- the init hook takes any number of JSON-safe arguments\n"
            "- the cleanup hook takes no arguments\n"
        ) from e

    return source_utils.resolve(func), validated_kwargs
Modules