lifecycle.FunctionInputOutputTypeChecker

Use this hook to print out data before/after a node’s execution for debugging

class hamilton.lifecycle.default.FunctionInputOutputTypeChecker(check_input: bool = True, check_output: bool = True)[source]

This lifecycle hook checks the input and output types of a function.

It is a simple, but very strict type check against the declared type with what was actually received. E.g. if you don’t want to check the types of a dictionary, don’t annotate it with a type.

__init__(check_input: bool = True, check_output: bool = True)[source]

Constructor.

Parameters:

• check_input – check inputs to all functions
• check_output – check outputs to all functions

post_node_execute(*, run_id: str, node_: Node, kwargs: dict[str, Any], success: bool, error: Exception | None, result: Any | None, task_id: str | None = None)

Wraps the after_execution method, providing a bridge to an external-facing API. Do not override this!

pre_node_execute(*, run_id: str, node_: Node, kwargs: dict[str, Any], task_id: str | None = None)

Wraps the before_execution method, providing a bridge to an external-facing API. Do not override this!

run_after_node_execution(node_name: str, node_tags: dict[str, Any], node_kwargs: dict[str, Any], node_return_type: type, result: Any, error: Exception | None, success: bool, task_id: str | None, run_id: str, **future_kwargs: Any)[source]

Checks that the result type matches the expected node return type.

run_before_node_execution(node_name: str, node_tags: dict[str, Any], node_kwargs: dict[str, Any], node_return_type: type, task_id: str | None, run_id: str, node_input_types: dict[str, Any], **future_kwargs: Any)[source]

Checks that the result type matches the expected node return type.