docstr: Return a list of TaskDataStore for a subset of the tasks.
  
  We filter the list based on `steps` if non-None.
  Alternatively, `pathspecs` can contain the exact list of pathspec(s)
  (run_id/step_name/task_id) that should be filtered.
  Note: When `pathspecs` is specified, we expect strict consistency and
  not eventual consistency in contrast to other modes.
  
  Parameters
  ----------
  run_id : str, optional
      Run ID to get the tasks from. If not specified, use pathspecs,
      by default None
  steps : List[str] , optional
      Steps to get the tasks from. If run_id is specified, this
      must also be specified, by default None
  pathspecs : List[str], optional
      Full task specs (run_id/step_name/task_id). Can be used instead of
      specifying run_id and steps, by default None
  allow_not_done : bool, optional
      If True, returns the latest attempt of a task even if that attempt
      wasn't marked as done, by default False
  
  Returns
  -------
  List[TaskDataStore]
      Task datastores for all the tasks specified.
docstr_tok: ["Return","a","list","of","TaskDataStore","for","a","subset","of","the","tasks",".","We","filter","the","list","based","on","`","steps","`","if","non-None",".","Alternatively",",","`","pathspecs","`","can","contain","the","exact","list","of","pathspec","(","s",")","(","run_id\/step_name\/task_id",")","that","should","be","filtered",".","Note",":","When","`","pathspecs","`","is","specified",",","we","expect","strict","consistency","and","not","eventual","consistency","in","contrast","to","other","modes",".","Parameters","--","--","--","--","--","run_id",":","str",",","optional","Run","ID","to","get","the","tasks","from",".","If","not","specified",",","use","pathspecs",",","by","default","None","steps",":","List","[","str","]",",","optional","Steps","to","get","the","tasks","from",".","If","run_id","is","specified",",","this","must","also","be","specified",",","by","default","None","pathspecs",":","List","[","str","]",",","optional","Full","task","specs","(","run_id\/step_name\/task_id",")",".","Can","be","used","instead","of","specifying","run_id","and","steps",",","by","default","None","allow_not_done",":","bool",",","optional","If","True",",","returns","the","latest","attempt","of","a","task","even","if","that","attempt","was","n't","marked","as","done",",","by","default","False","Returns","--","--","--","-","List","[","TaskDataStore","]","Task","datastores","for","all","the","tasks","specified","."]
code: def get_latest_task_datastores(
      self,
      run_id=None,
      steps=None,
      pathspecs=None,
      allow_not_done=False,
  ):
      """
      Return a list of TaskDataStore for a subset of the tasks.
  
      We filter the list based on `steps` if non-None.
      Alternatively, `pathspecs` can contain the exact list of pathspec(s)
      (run_id/step_name/task_id) that should be filtered.
      Note: When `pathspecs` is specified, we expect strict consistency and
      not eventual consistency in contrast to other modes.
  
      Parameters
      ----------
      run_id : str, optional
          Run ID to get the tasks from. If not specified, use pathspecs,
          by default None
      steps : List[str] , optional
          Steps to get the tasks from. If run_id is specified, this
          must also be specified, by default None
      pathspecs : List[str], optional
          Full task specs (run_id/step_name/task_id). Can be used instead of
          specifying run_id and steps, by default None
      allow_not_done : bool, optional
          If True, returns the latest attempt of a task even if that attempt
          wasn't marked as done, by default False
  
      Returns
      -------
      List[TaskDataStore]
          Task datastores for all the tasks specified.
      """
      task_urls = []
      # Note: When `pathspecs` is specified, we avoid the potentially
      # eventually consistent `list_content` operation, and directly construct
      # the task_urls list.
      if pathspecs:
          task_urls = [
              self._storage_impl.path_join(self.flow_name, pathspec)
              for pathspec in pathspecs
          ]
      else:
          run_prefix = self._storage_impl.path_join(
              self.flow_name, run_id
          )
          if steps:
              step_urls = [
                  self._storage_impl.path_join(run_prefix, step)
                  for step in steps
              ]
          else:
              step_urls = [
                  step.path
                  for step in self._storage_impl.list_content(
                      [run_prefix]
                  )
                  if step.is_file is False
              ]
          task_urls = [
              task.path
              for task in self._storage_impl.list_content(step_urls)
              if task.is_file is False
          ]
      urls = []
      for task_url in task_urls:
          for attempt in range(metaflow_config.MAX_ATTEMPTS):
              for suffix in [
                  TaskDataStore.METADATA_DATA_SUFFIX,
                  TaskDataStore.METADATA_ATTEMPT_SUFFIX,
                  TaskDataStore.METADATA_DONE_SUFFIX,
              ]:
                  urls.append(
                      self._storage_impl.path_join(
                          task_url,
                          TaskDataStore.metadata_name_for_attempt(
                              suffix, attempt
                          ),
                      )
                  )
  
      latest_started_attempts = {}
      done_attempts = set()
      data_objs = {}
      with self._storage_impl.load_bytes(urls) as get_results:
          for key, path, meta in get_results:
              if path is not None:
                  (
                      _,
                      run,
                      step,
                      task,
                      fname,
                  ) = self._storage_impl.path_split(key)
                  attempt, fname = TaskDataStore.parse_attempt_metadata(
                      fname
                  )
                  attempt = int(attempt)
                  if fname == TaskDataStore.METADATA_DONE_SUFFIX:
                      done_attempts.add((run, step, task, attempt))
                  elif fname == TaskDataStore.METADATA_ATTEMPT_SUFFIX:
                      latest_started_attempts[(run, step, task)] = max(
                          latest_started_attempts.get(
                              (run, step, task), 0
                          ),
                          attempt,
                      )
                  elif fname == TaskDataStore.METADATA_DATA_SUFFIX:
                      # This somewhat breaks the abstraction since we are using
                      # load_bytes directly instead of load_metadata
                      with open(path, encoding="utf-8") as f:
                          data_objs[
                              (run, step, task, attempt)
                          ] = json.load(f)
      # We now figure out the latest attempt that started *and* finished.
      # Note that if an attempt started but didn't finish, we do *NOT* return
      # the previous attempt
      latest_started_attempts = set(
          (run, step, task, attempt)
          for (
              run,
              step,
              task,
          ), attempt in latest_started_attempts.items()
      )
      if allow_not_done:
          latest_to_fetch = latest_started_attempts
      else:
          latest_to_fetch = latest_started_attempts & done_attempts
      latest_to_fetch = [
          (
              v[0],
              v[1],
              v[2],
              v[3],
              data_objs.get(v),
              "r",
              allow_not_done,
          )
          for v in latest_to_fetch
      ]
      return list(
          itertools.starmap(self.get_task_datastore, latest_to_fetch)
      )
  
file_path: metaflow/metaflow/datastore/flow_datastore.py
imports: import itertools
  import json
  from ..None import metaflow_config
  from .content_addressed_store import ContentAddressedStore
  from .task_datastore import TaskDataStore
label_desc: Use image node_id 3 for calling the FlowDataStore obj's underlying member method code with example usage: obj.get_latest_task_datastores(run_id, steps, pathspecs, allow_not_done) and returns: list