class_docstr: Allows previously-generated Artifacts to be used in a new pipeline run.
  
  By implementing this in a class (instead of a function), we reduce the
  number of MLMD reads when reusing the outputs of multiple nodes in the same
  pipeline run.
class_docstr_tok: ["Allows","previously-generated","Artifacts","to","be","used","in","a","new","pipeline","run",".","By","implementing","this","in","a","class","(","instead","of","a","function",")",",","we","reduce","the","number","of","MLMD","reads","when","reusing","the","outputs","of","multiple","nodes","in","the","same","pipeline","run","."]
docstr: Gets the latest previous pipeline_run_id.
docstr_tok: ["Gets","the","latest","previous","pipeline_run_id","."]
code: def _get_base_pipeline_run_context(
      self, base_run_id: Optional[str] = None
  ) -> metadata_store_pb2.Context:
      """Gets the latest previous pipeline_run_id."""
      child_contexts = (
          self._mlmd.store.get_children_contexts_by_context(
              self._pipeline_context.id
          )
      )
      if child_contexts:
          pipeline_run_type_id = self._mlmd.store.get_context_type(
              constants.PIPELINE_RUN_CONTEXT_TYPE_NAME
          ).id
          pipeline_run_contexts = {
              ctx.name: ctx
              for ctx in child_contexts
              if ctx.type_id == pipeline_run_type_id
          }
      else:
          logging.info(
              "No child contexts found. Falling back to previous logic."
          )
          # The parent-child relationship between pipeline and pipeline run contexts
          # is set up after the partial run feature is available to users. For
          # existing pipelines, we need to fall back to the previous logic.
          pipeline_run_contexts = {
              run_ctx.name: run_ctx
              for run_ctx in self._mlmd.store.get_contexts_by_type(
                  constants.PIPELINE_RUN_CONTEXT_TYPE_NAME
              )
          }
      if base_run_id:
          if base_run_id in pipeline_run_contexts:
              return pipeline_run_contexts[base_run_id]
          else:
              raise LookupError(
                  f"pipeline_run_id {base_run_id} not found in MLMD."
              )
  
      pipeline_run_contexts = [
          c
          for c in pipeline_run_contexts.values()
          if c.name != self._new_run_id
      ]
      sorted_run_contexts = sorted(
          pipeline_run_contexts, key=lambda c: c.create_time_since_epoch
      )
      if not sorted_run_contexts:
          return None
  
      logging.info(
          "base_run_id not provided. Default to latest pipeline run: %s",
          sorted_run_contexts[-1].name,
      )
      return sorted_run_contexts[-1]
  
imports: import collections
  import concurrent.futures
  import enum
  from typing import Collection, Dict, Final, List, Mapping, Optional, OrderedDict, Sequence, Set, Tuple
  import uuid
  from absl import logging
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.compiler import constants
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 2 for calling the _ArtifactRecycler obj's underlying member method code with example usage: obj._get_base_pipeline_run_context(base_run_id) and returns: sorted_run_contexts, None, pipeline_run_contexts