lifecycle.api.StaticValidator

class hamilton.lifecycle.api.StaticValidator[source]

Performs static validation of the DAG. Note that this has the option to perform default validation for each method – this means that if you don’t implement one of these it is OK.

class MyTagValidator(api.StaticValidator):
    '''Validates tags on a node'''

    def run_to_validate_node(
            self, *, node: HamiltonNode, **future_kwargs
    ) -> tuple[bool, Optional[str]]:
        if node.tags.get("node_type", "") == "output":
            table_name = node.tags.get("table_name")
            if not table_name:  # None or empty
                error_msg = (f"Node {node.tags['module']}.{node.name} "
                            "is an output node, but does not have a table_name tag.")
                return False, error_msg
        return True, None

run_to_validate_graph(graph: HamiltonGraph, **future_kwargs) → tuple[bool, str | None][source]

Override this to build custom DAG validations! Default to just returning that the graph is valid, so you don’t have to implement it if you want to just implement a single method. Runs post graph construction to validate a graph. You have access to a bunch of metadata about the graph, stored in the graph argument.

Parameters:

• graph – Graph to validate.
• future_kwargs – Additional keyword arguments – this is kept for backwards compatibility

Returns:

A tuple of whether the graph is valid and an error message in the case of failure. Return [True, None] for a valid graph. Otherwise, return a detailed error message – this should have all context/debugging information.

run_to_validate_node(*, node: HamiltonNode, **future_kwargs) → tuple[bool, str | None][source]

Override this to build custom node validations! Defaults to just returning that a node is valid so you don’t have to implement it if you want to just implement a single method. Runs post node construction to validate a node. You have access to a bunch of metadata about the node, stored in the hamilton_node argument

Parameters:

• node – Node to validate
• future_kwargs – Additional keyword arguments – this is kept for backwards compatibility

Returns:

A tuple of whether the node is valid and an error message in the case of failure. Return [True, None] for a valid node.Otherwise, return a detailed error message – this should have all context/debugging information, but does not need to mention the node name (it will be aggregated with others).

final validate_graph(*, graph: FunctionGraph, modules: list[ModuleType], config: dict[str, Any]) → tuple[bool, Exception | None][source]

Validates the graph. This will raise an InvalidNodeException

Parameters:

• graph – Graph that has been constructed.
• modules – Modules passed into the graph
• config – Config passed into the graph

Returns:

A (is_valid, error_message) tuple

final validate_node(*, created_node: Node) → tuple[bool, Exception | None][source]

Validates a node. This will raise an InvalidNodeException if the node is invalid.

Parameters:

created_node – Node that was created.

Raises:

InvalidNodeException – If the node is invalid.