class_docstr: Helper class to handle metadata I/O.
  
  Not thread-safe without external synchronisation.
docstr: Gets execution type given execution type name and properties.
  
  Uses existing type if schema is superset of what is needed. Otherwise tries
  to register new execution type.
  
  Args:
    type_name: the name of the execution type
    exec_properties: the execution properties included by the execution
  
  Returns:
    execution type id
  Raises:
    ValueError if new execution type conflicts with existing schema in MLMD.
docstr_tok: ["Gets","execution","type","given","execution","type","name","and","properties",".","Uses","existing","type","if","schema","is","superset","of","what","is","needed",".","Otherwise","tries","to","register","new","execution","type",".","Args",":","type_name",":","the","name","of","the","execution","type","exec_properties",":","the","execution","properties","included","by","the","execution","Returns",":","execution","type","id","Raises",":","ValueError","if","new","execution","type","conflicts","with","existing","schema","in","MLMD","."]
code: def _prepare_execution_type(
      self, type_name: str, exec_properties: Dict[str, Any]
  ) -> int:
      """Gets execution type given execution type name and properties.
  
      Uses existing type if schema is superset of what is needed. Otherwise tries
      to register new execution type.
  
      Args:
        type_name: the name of the execution type
        exec_properties: the execution properties included by the execution
  
      Returns:
        execution type id
      Raises:
        ValueError if new execution type conflicts with existing schema in MLMD.
      """
      existing_execution_type = None
      try:
          existing_execution_type = self.store.get_execution_type(
              type_name
          )
          if existing_execution_type is None:
              raise RuntimeError(
                  "Execution type is None for %s." % type_name
              )
          # If exec_properties contains new entries, execution type schema will be
          # updated in MLMD.
          if all(
              k in existing_execution_type.properties
              for k in exec_properties.keys()
          ):
              return existing_execution_type.id
          else:
              raise mlmd.errors.NotFoundError(
                  "No qualified execution type found."
              )
      except mlmd.errors.NotFoundError:
          execution_type = metadata_store_pb2.ExecutionType(
              name=type_name
          )
          execution_type.properties[
              _EXECUTION_TYPE_KEY_STATE
          ] = metadata_store_pb2.STRING
          # TODO(b/172673657): Make property schema registration more explicit to
          # the component author.
          for k in exec_properties.keys():
              assert k not in _EXECUTION_TYPE_RESERVED_KEYS, (
                  "execution properties with reserved key %s"
              ) % k
              # TODO(b/172629873): Currently, all execution properties are registered
              # as strings.
              execution_type.properties[k] = metadata_store_pb2.STRING
          # TODO(ruoyu): Find a better place / solution to the checksum logic.
          if "module_file" in exec_properties:
              execution_type.properties[
                  _EXECUTION_TYPE_KEY_CHECKSUM
              ] = metadata_store_pb2.STRING
          execution_type.properties[
              _EXECUTION_TYPE_KEY_PIPELINE_NAME
          ] = metadata_store_pb2.STRING
          execution_type.properties[
              _EXECUTION_TYPE_KEY_PIPELINE_ROOT
          ] = metadata_store_pb2.STRING
          execution_type.properties[
              _EXECUTION_TYPE_KEY_RUN_ID
          ] = metadata_store_pb2.STRING
          execution_type.properties[
              _EXECUTION_TYPE_KEY_COMPONENT_ID
          ] = metadata_store_pb2.STRING
  
          try:
              # As exec_properties infers an execution_type dynamically. The stored
              # type may have more properties which are inferred from other instances
              # and the current exec_properties may introduce new keys at run time.
              # Here we enable `can_add_fields` and `can_omit_fields` so that the
              # stored execution_type are evolved to union all keys set by different
              # exec_properties instances.
              execution_type_id = self.store.put_execution_type(
                  execution_type=execution_type,
                  can_add_fields=True,
                  can_omit_fields=True,
              )
              absl.logging.debug(
                  "Registering an execution type with id %s."
                  % execution_type_id
              )
              return execution_type_id
          except mlmd.errors.AlreadyExistsError:
              # The conflict should not happen as all property value type is STRING.
              warning_str = (
                  "Conflicting properties in exec_properties comparing with "
                  "existing execution type with the same type name. Existing type: "
                  "%s, New type: %s"
              ) % (existing_execution_type, execution_type)
              absl.logging.warning(warning_str)
              raise ValueError(warning_str)
  
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 15 for calling the Metadata obj's underlying member method code with example usage: obj._prepare_execution_type(type_name, exec_properties) and returns: existing_execution_type, execution_type_id