class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Registers a new execution in metadata.
  
  Args:
    pipeline_info: optional pipeline info of the execution.
    component_info: optional component info of the execution.
    contexts: contexts for current run, all contexts will be linked to the
      execution. In addition, a component run context will be added to the
      contexts list.
    exec_properties: the execution properties of the execution.
    input_artifacts: input artifacts of the execution.
  
  Returns:
    execution id of the new execution.
docstr_tok: ["Registers","a","new","execution","in","metadata",".","Args",":","pipeline_info",":","optional","pipeline","info","of","the","execution",".","component_info",":","optional","component","info","of","the","execution",".","contexts",":","contexts","for","current","run",",","all","contexts","will","be","linked","to","the","execution",".","In","addition",",","a","component","run","context","will","be","added","to","the","contexts","list",".","exec_properties",":","the","execution","properties","of","the","execution",".","input_artifacts",":","input","artifacts","of","the","execution",".","Returns",":","execution","id","of","the","new","execution","."]
code: def register_execution(
      self,
      pipeline_info: data_types.PipelineInfo,
      component_info: data_types.ComponentInfo,
      contexts: List[metadata_store_pb2.Context],
      exec_properties: Optional[Dict[str, Any]] = None,
      input_artifacts: Optional[Dict[str, List[_Artifact]]] = None,
  ) -> metadata_store_pb2.Execution:
      """Registers a new execution in metadata.
  
      Args:
        pipeline_info: optional pipeline info of the execution.
        component_info: optional component info of the execution.
        contexts: contexts for current run, all contexts will be linked to the
          execution. In addition, a component run context will be added to the
          contexts list.
        exec_properties: the execution properties of the execution.
        input_artifacts: input artifacts of the execution.
  
      Returns:
        execution id of the new execution.
      """
      input_artifacts = input_artifacts or {}
      exec_properties = exec_properties or {}
      execution = self._prepare_execution(
          EXECUTION_STATE_NEW,
          exec_properties,
          pipeline_info,
          component_info,
      )
      artifacts_and_events = self._artifact_and_event_pairs(
          artifact_dict=input_artifacts,
          event_type=metadata_store_pb2.Event.INPUT,
      )
      component_run_context = self._prepare_context(
          context_type_name=_CONTEXT_TYPE_COMPONENT_RUN,
          context_name=component_info.component_run_context_name,
          properties={
              _CONTEXT_TYPE_KEY_PIPELINE_NAME: pipeline_info.pipeline_name,
              _CONTEXT_TYPE_KEY_RUN_ID: pipeline_info.run_id,
              _CONTEXT_TYPE_KEY_COMPONENT_ID: component_info.component_id,
          },
      )
      # Tries to register the execution along with a component run context. If the
      # context already exists, reuse the context and update the existing
      # execution.
      try:
          execution_id, a_ids, context_ids = self.store.put_execution(
              execution=execution,
              artifact_and_events=artifacts_and_events,
              contexts=contexts + [component_run_context],
          )
          execution.id = execution_id
          component_run_context.id = context_ids[-1]
      except mlmd.errors.AlreadyExistsError:
          component_run_context = self.get_component_run_context(
              component_info
          )
          assert (
              component_run_context is not None
          )  # AlreadyExistsError indicates.
          absl.logging.debug(
              "Component run context already exists. Reusing the context %s.",
              component_run_context.name,
          )
          [previous_execution] = self.store.get_executions_by_context(
              context_id=component_run_context.id
          )
          execution.id = previous_execution.id
          _, a_ids, _ = self.store.put_execution(
              execution=execution,
              artifact_and_events=artifacts_and_events,
              contexts=contexts + [component_run_context],
          )
      contexts.append(component_run_context)
      for artifact_and_event, a_id in zip(artifacts_and_events, a_ids):
          artifact_and_event[0].id = a_id
      return execution
  
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 20 for calling the Metadata obj's underlying member method code with example usage: obj.register_execution(pipeline_info, component_info, contexts, exec_properties, input_artifacts) and returns: execution