code: def _explore_from_execution(
      starting_execution_id: int,
      direction: metadata_store_pb2.LineageSubgraphQueryOptions.Direction,
      resolver_graph: ResolverGraph,
      visited_ids: Dict[NodeType, Set[int]],
      subgraph: metadata_store_pb2.LineageGraph,
  ):
      """Given a starting execution, runs a single dfs on the graph from it.
  
      Args:
        starting_execution_id: starting execution id.
        direction: direction of dfs. It can be single-directional or bidirectional.
        resolver_graph: resolver graph representing the lineage graph to run dfs on.
        visited_ids: a set of visited node ids.
        subgraph: lineage graph to store returned nodes from dfs.
      """
      visited_ids[NodeType.EXECUTION].add(starting_execution_id)
      # If no executions are returned with the lineage_graph from
      # `get_lineage_subgraph()`, the `resolver_graph` will also have
      # `executions_by_id` being empty. Therefore we don't append any execution to
      # the returned `subgraph`.
      if resolver_graph.executions_by_id:
          subgraph.executions.append(
              resolver_graph.executions_by_id[starting_execution_id]
          )
      if direction in [
          metadata_store_pb2.LineageSubgraphQueryOptions.Direction.UPSTREAM,
          metadata_store_pb2.LineageSubgraphQueryOptions.Direction.BIDIRECTIONAL,
      ]:
          if (
              starting_execution_id
              in resolver_graph.execution_to_artifact[
                  EventType.INPUT
              ].keys()
          ):
              for artifact_id in resolver_graph.execution_to_artifact[
                  EventType.INPUT
              ][starting_execution_id]:
                  if artifact_id not in visited_ids[NodeType.ARTIFACT]:
                      _explore_from_artifact(
                          artifact_id,
                          direction,
                          resolver_graph,
                          visited_ids,
                          subgraph,
                      )
      if direction in [
          metadata_store_pb2.LineageSubgraphQueryOptions.Direction.DOWNSTREAM,
          metadata_store_pb2.LineageSubgraphQueryOptions.Direction.BIDIRECTIONAL,
      ]:
          if (
              starting_execution_id
              in resolver_graph.execution_to_artifact[
                  EventType.OUTPUT
              ].keys()
          ):
              for artifact_id in resolver_graph.execution_to_artifact[
                  EventType.OUTPUT
              ][starting_execution_id]:
                  if artifact_id not in visited_ids[NodeType.ARTIFACT]:
                      _explore_from_artifact(
                          artifact_id,
                          direction,
                          resolver_graph,
                          visited_ids,
                          subgraph,
                      )
  
file_path: ml-metadata/ml_metadata/tools/mlmd_resolver/metadata_resolver_utils.py
imports: import collections
  import enum
  from typing import Callable, Dict, List, Optional, Set
  import attr
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 4 for calling a global function with example usage: _explore_from_execution(starting_execution_id, direction, resolver_graph, visited_ids, subgraph) without return types