code: def get_cache_context(
      metadata_handle: metadata.Metadata,
      pipeline_node: pipeline_pb2.PipelineNode,
      pipeline_info: pipeline_pb2.PipelineInfo,
      executor_spec: Optional[message.Message] = None,
      input_artifacts: Optional[
          Mapping[str, Sequence[types.Artifact]]
      ] = None,
      output_artifacts: Optional[
          Mapping[str, Sequence[types.Artifact]]
      ] = None,
      parameters: Optional[Mapping[str, Any]] = None,
  ) -> metadata_store_pb2.Context:
      """Gets cache context for a potential node execution.
  
      The cache key is generated by applying SHA-256 hashing function on:
      - Serialized pipeline info.
      - Serialized node_info of the PipelineNode.
      - Serialized executor spec
      - Serialized input artifacts if any.
      - Serialized output artifacts if any. The uri was removed during the process.
      - Serialized parameters if any.
      - Serialized module file content if module file is present in parameters.
  
      Args:
        metadata_handle: A handler to access MLMD store.
        pipeline_node: A pipeline_pb2.PipelineNode instance to represent the node.
        pipeline_info: Information of the pipeline.
        executor_spec: A proto message representing the executor specification.
        input_artifacts: Input artifacts of the potential execution. The order of
          the artifacts under a key matters when calculating the cache key.
        output_artifacts: Output artifacts skeleton of the potential execution. The
          order of the artifadts under a key matters when calculating the cache key.
        parameters: Parameters of the potential execution.
  
      Returns:
        A metadata_store_pb2.Context for the cache key.
      """
      h = hashlib.sha256()
      h.update(pipeline_info.SerializeToString(deterministic=True))
      h.update(
          pipeline_node.node_info.SerializeToString(deterministic=True)
      )
      if executor_spec:
          h.update(executor_spec.SerializeToString(deterministic=True))
      for key in sorted(input_artifacts or {}):
          h.update(key.encode())
          for artifact in input_artifacts[key]:
              h.update(
                  artifact.mlmd_artifact.SerializeToString(
                      deterministic=True
                  )
              )
      for key in sorted(output_artifacts or {}):
          h.update(key.encode())
          for artifact in output_artifacts[key]:
              stateless_artifact = copy.deepcopy(artifact)
              # Output uri and name should not be taken into consideration as cache key.
              stateless_artifact.uri = ""
              stateless_artifact.name = ""
              h.update(
                  stateless_artifact.mlmd_artifact.SerializeToString(
                      deterministic=True
                  )
              )
      parameters = parameters or {}
      for key, value in sorted(parameters.items()):
          h.update(key.encode())
          h.update(str(value).encode())
  
      # Special treatment for module files as they will be used as part of the logic
      # for processing. Currently this pattern is employeed by Trainer and
      # Transform.
      if (
          "module_file" in parameters
          and parameters["module_file"]
          and fileio.exists(parameters["module_file"])
      ):
          with fileio.open(parameters["module_file"], "r") as f:
              h.update(f.read().encode())
  
      return context_lib.register_context_if_not_exists(
          metadata_handle=metadata_handle,
          context_type_name=context_lib.CONTEXT_TYPE_EXECUTION_CACHE,
          context_name=h.hexdigest(),
      )
  
file_path: tfx/tfx/orchestration/portable/cache_utils.py
imports: import collections
  import copy
  import hashlib
  from typing import Any, Dict, List, Mapping, Optional, Sequence
  from tfx import types
  from tfx.dsl.io import fileio
  from tfx.orchestration import metadata
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import artifact_utils
  from google.protobuf import message
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 2 for calling a global function with example usage: get_cache_context(metadata_handle, pipeline_node, pipeline_info, executor_spec, input_artifacts, output_artifacts, parameters) and returns: context_lib