class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Publishes an execution with input and output artifacts info.
  
  This method will publish any execution with non-final states. It will
  register unseen artifacts and publish events for them.
  
  Args:
    component_info: component information.
    output_artifacts: output artifacts produced by the execution.
    exec_properties: execution properties for the execution to be published.
docstr_tok: ["Publishes","an","execution","with","input","and","output","artifacts","info",".","This","method","will","publish","any","execution","with","non-final","states",".","It","will","register","unseen","artifacts","and","publish","events","for","them",".","Args",":","component_info",":","component","information",".","output_artifacts",":","output","artifacts","produced","by","the","execution",".","exec_properties",":","execution","properties","for","the","execution","to","be","published","."]
code: def publish_execution(
      self,
      component_info: data_types.ComponentInfo,
      output_artifacts: Optional[Dict[str, List[_Artifact]]] = None,
      exec_properties: Optional[Dict[str, Any]] = None,
  ) -> None:
      """Publishes an execution with input and output artifacts info.
  
      This method will publish any execution with non-final states. It will
      register unseen artifacts and publish events for them.
  
      Args:
        component_info: component information.
        output_artifacts: output artifacts produced by the execution.
        exec_properties: execution properties for the execution to be published.
      """
      component_run_context = self.get_component_run_context(
          component_info
      )
      if component_run_context is None:
          raise ValueError("Component run context does not exist.")
      [execution] = self.store.get_executions_by_context(
          component_run_context.id
      )
      contexts = [
          component_run_context,
          self.get_pipeline_run_context(component_info.pipeline_info),
          self.get_pipeline_context(component_info.pipeline_info),
      ]
      contexts = [ctx for ctx in contexts if ctx is not None]
      # If execution state is already in final state, skips publishing.
      if (
          execution.properties[_EXECUTION_TYPE_KEY_STATE].string_value
          in FINAL_EXECUTION_STATES
      ):
          return
      self.update_execution(
          execution=execution,
          component_info=component_info,
          output_artifacts=output_artifacts,
          exec_properties=exec_properties,
          execution_state=EXECUTION_STATE_COMPLETE,
          artifact_state=_ArtifactState.PUBLISHED,
          contexts=contexts,
      )
  
imports: import collections
  import copy
  import hashlib
  import itertools
  import os
  import random
  import time
  import types
  from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union
  import absl
  from tfx.dsl.io import fileio
  from tfx.orchestration import data_types
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.types import artifact
  from tfx.types import artifact_utils
  import ml_metadata
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 21 for calling the Metadata obj's underlying member method code with example usage: obj.publish_execution(component_info, output_artifacts, exec_properties) without return types