code: def put_executions(
      metadata_handle: metadata.Metadata,
      executions: Sequence[metadata_store_pb2.Execution],
      contexts: Sequence[metadata_store_pb2.Context],
      input_artifacts_maps: Optional[
          Sequence[typing_utils.ArtifactMultiMap]
      ] = None,
      output_artifacts_maps: Optional[
          Sequence[typing_utils.ArtifactMultiMap]
      ] = None,
      input_event_type: metadata_store_pb2.Event.Type = metadata_store_pb2.Event.INPUT,
      output_event_type: metadata_store_pb2.Event.Type = metadata_store_pb2.Event.OUTPUT,
  ) -> Sequence[metadata_store_pb2.Execution]:
      """Writes an execution-centric subgraph to MLMD.
  
      This function mainly leverages metadata.put_lineage_subgraph() method to write
      the execution centric subgraph to MLMD.
  
      Args:
        metadata_handle: A handler to access MLMD.
        executions: A list of executions to be written to MLMD.
        contexts: A list of MLMD contexts to associated with all the executions.
        input_artifacts_maps: A list of ArtifactMultiMap for input. Each of the
          ArtifactMultiMap links with one execution. None or empty
          input_artifacts_maps means no input for the executions.
        output_artifacts_maps: A list of ArtifactMultiMap for output. Each of the
          ArtifactMultiMap links with one execution. None or empty
          output_artifacts_maps means no input for the executions.
        input_event_type: The type of the input event, default to be INPUT.
        output_event_type: The type of the output event, default to be OUTPUT.
  
      Returns:
        A list of MLMD executions that are written to MLMD, with id pupulated.
      """
      if input_artifacts_maps and len(executions) != len(
          input_artifacts_maps
      ):
          raise ValueError(
              f"The number of executions {len(executions)} should be the same as "
              f"the number of input ArtifactMultiMap {len(input_artifacts_maps)}."
          )
      if output_artifacts_maps and len(executions) != len(
          output_artifacts_maps
      ):
          raise ValueError(
              f"The number of executions {len(executions)} should be the same as "
              f"the number of output ArtifactMultiMap {len(output_artifacts_maps)}."
          )
  
      artifacts = []
      input_artifact_idx_by_id = {}
      artifact_event_edges = []
      if input_artifacts_maps:
          for idx, input_artifacts in enumerate(input_artifacts_maps):
              artifact_and_event_pairs = (
                  _create_artifact_and_event_pairs(
                      metadata_handle,
                      input_artifacts,
                      event_type=input_event_type,
                  )
              )
              for artifact, event in artifact_and_event_pairs:
                  artifact_id = _get_id(artifact)
                  if not artifact_id:
                      artifacts.append(artifact)
                      artifact_event_edges.append(
                          (idx, len(artifacts) - 1, event)
                      )
                  elif artifact_id in input_artifact_idx_by_id:
                      artifact_event_edges.append(
                          (
                              idx,
                              input_artifact_idx_by_id[artifact_id],
                              event,
                          )
                      )
                  else:
                      artifacts.append(artifact)
                      artifact_event_edges.append(
                          (idx, len(artifacts) - 1, event)
                      )
                      input_artifact_idx_by_id[artifact_id] = (
                          len(artifacts) - 1
                      )
  
      if output_artifacts_maps:
          for idx, output_artifacts in enumerate(output_artifacts_maps):
              outputs_utils.tag_output_artifacts_with_version(
                  output_artifacts
              )
              artifact_and_event_pairs = (
                  _create_artifact_and_event_pairs(
                      metadata_handle,
                      output_artifacts,
                      event_type=output_event_type,
                  )
              )
              for artifact, event in artifact_and_event_pairs:
                  artifacts.append(artifact)
                  artifact_event_edges.append(
                      (idx, len(artifacts) - 1, event)
                  )
  
      try:
          (
              execution_ids,
              artifact_ids,
              context_ids,
          ) = metadata_handle.store.put_lineage_subgraph(
              executions,
              artifacts,
              contexts,
              artifact_event_edges,
              reuse_context_if_already_exist=True,
              reuse_artifact_if_already_exist_by_external_id=True,
          )
          for execution, execution_id in zip(executions, execution_ids):
              execution.id = execution_id
          for artifact, artifact_id in zip(artifacts, artifact_ids):
              artifact.id = artifact_id
          for context, context_id in zip(contexts, context_ids):
              context.id = context_id
          return executions
      except mlmd.errors.UnimplementedError:
          logging.warning(
              "PutLineageSubgraph API do not exist. Falling back to PutExecution."
          )
          result = []
          for (
              execution,
              input_artifacts,
              output_artifacts,
          ) in itertools.zip_longest(
              executions,
              input_artifacts_maps or [],
              output_artifacts_maps or [],
          ):
              result.append(
                  put_execution(
                      metadata_handle=metadata_handle,
                      execution=execution,
                      contexts=contexts,
                      input_artifacts=input_artifacts,
                      output_artifacts=output_artifacts,
                      input_event_type=input_event_type,
                      output_event_type=output_event_type,
                  )
              )
          return result
  
imports: from __future__ import annotations
  import collections
  import copy
  import itertools
  import re
  from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
  from absl import logging
  from tfx import types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import artifact_lib
  from tfx.orchestration.portable.mlmd import common_utils
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.orchestration.portable.mlmd import filter_query_builder
  from tfx.utils import metrics_utils
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import proto_utils
  from tfx.utils import typing_utils
  from tfx.utils import telemetry_utils
  from google.protobuf import json_format
  import ml_metadata
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 15 for calling a global function with example usage: put_executions(metadata_handle, executions, contexts, input_artifacts_maps, output_artifacts_maps, input_event_type, output_event_type) and returns: executions, result