hamilton_experiment_hook_pseudo.py

class ExperimentTracker(
    lifecycle.NodeExecutionHook,
    lifecycle.GraphExecutionHook,
    lifecycle.GraphConstructionHook,
):
    def __init__(self, experiment_name: str, base_directory: str):
        self.cache = Cache(base_directory)
        self.init_directory = Path.cwd()
        self.run_directory = resolve_path(base_directory)
        self.materialized = list()

    def post_graph_construct(self, config):
        """Collect the Driver config"""
        self.config = config

    def run_before_graph_execution(self, run_id, graph, inputs, overrides):
        """Gather metadata about the starting run"""
        self.run_id = run_id
        self.graph_hash = hash(graph)
        self.inputs = get_input_nodes(graph.nodes)
        self.overrides = overrides
        
    def run_before_node_execution(self, node):
        """Move to run directory to produce materialized result"""
        if is_materializer(node):
            os.chdir(self.run_directory)

    def run_after_node_execution(self, node):
        """Return to init directory after materialization. Log materialization"""
        os.chdir(self.init_directory)
        self.materializers.append(node)

    def run_after_graph_execution(self):
        """Collect metadata stored in tracker and write to cache"""
        metadata = get_run_metadata(self)
        self.cache.write(self.run_id, metadata)