code: def _get_outputs_of_execution(
      metadata_handle: metadata.Metadata, execution_id: int
  ) -> Optional[Dict[str, List[types.Artifact]]]:
      """Fetches outputs produced by a historical execution.
  
      Args:
        metadata_handle: A handler to access MLMD store.
        execution_id: The id of the execution that produced the outputs.
  
      Returns:
        A dict of key -> List[Artifact] as the result if qualified outputs found.
        Otherwise returns None.
      """
      result = collections.defaultdict(list)
  
      output_events = [
          event
          for event in metadata_handle.store.get_events_by_execution_ids(
              [execution_id]
          )
          if event.type == metadata_store_pb2.Event.OUTPUT
      ]
      cached_output_artifacts = (
          metadata_handle.store.get_artifacts_by_id(
              [e.artifact_id for e in output_events]
          )
      )
      for artifact in cached_output_artifacts:
          # Non-live artifact means partial result, will not be used.
          if artifact.state != metadata_store_pb2.Artifact.LIVE:
              return None
  
      artifact_types = metadata_handle.store.get_artifact_types_by_id(
          [a.type_id for a in cached_output_artifacts]
      )
  
      for event, mlmd_artifact, artifact_type in zip(
          output_events, cached_output_artifacts, artifact_types
      ):
          key = event.path.steps[0].key
          tfx_artifact = artifact_utils.deserialize_artifact(
              artifact_type, mlmd_artifact
          )
          result[key].append(tfx_artifact)
  
      return result
  
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 1 for calling a global function with example usage: _get_outputs_of_execution(metadata_handle, execution_id) and returns: result, None