class_docstr: TaskDataStore is obtained through FlowDataStore.get_datastore_for_task and
  is used to store three things:
      - Task artifacts (using save_artifacts and load_artifacts) which will
        ultimately be stored using ContentAddressedStore's save_blobs and
        load_blobs. This is basically the content indexed portion of the
        storage (identical objects are stored only once).
      - Metadata information (using save_metadata and load_metadata) which
        stores JSON encoded metadata about a task in a non-content indexed
        way in a hierarchical manner (ie: the files are stored
        in a path indicated by the pathspec (run_id/step_name/task_id)).
        This portion of the store can be viewed as name indexed (storing
        two metadata items with the same name will overwrite the previous item
        so the condition of equality is the name as
        opposed to the content).
      - Logs which are a special sort of task metadata but are handled
        differently (they are not JSON-encodable dictionaries).
class_docstr_tok: ["TaskDataStore","is","obtained","through","FlowDataStore.get_datastore_for_task","and","is","used","to","store","three","things",":","-","Task","artifacts","(","using","save_artifacts","and","load_artifacts",")","which","will","ultimately","be","stored","using","ContentAddressedStore","'s","save_blobs","and","load_blobs",".","This","is","basically","the","content","indexed","portion","of","the","storage","(","identical","objects","are","stored","only","once",")",".","-","Metadata","information","(","using","save_metadata","and","load_metadata",")","which","stores","JSON","encoded","metadata","about","a","task","in","a","non-content","indexed","way","in","a","hierarchical","manner","(","ie",":","the","files","are","stored","in","a","path","indicated","by","the","pathspec","(","run_id\/step_name\/task_id",")",")",".","This","portion","of","the","store","can","be","viewed","as","name","indexed","(","storing","two","metadata","items","with","the","same","name","will","overwrite","the","previous","item","so","the","condition","of","equality","is","the","name","as","opposed","to","the","content",")",".","-","Logs","which","are","a","special","sort","of","task","metadata","but","are","handled","differently","(","they","are","not","JSON-encodable","dictionaries",")","."]
code: def __init__(
      self,
      flow_datastore,
      run_id,
      step_name,
      task_id,
      attempt=None,
      data_metadata=None,
      mode="r",
      allow_not_done=False,
  ):
      self._storage_impl = flow_datastore._storage_impl
      self.TYPE = self._storage_impl.TYPE
      self._ca_store = flow_datastore.ca_store
      self._environment = flow_datastore.environment
      self._run_id = run_id
      self._step_name = step_name
      self._task_id = task_id
      self._path = self._storage_impl.path_join(
          flow_datastore.flow_name, run_id, step_name, task_id
      )
      self._mode = mode
      self._attempt = attempt
      self._metadata = flow_datastore.metadata
      self._parent = flow_datastore
  
      # The GZIP encodings are for backward compatibility
      self._encodings = {"pickle-v2", "gzip+pickle-v2"}
      ver = sys.version_info[0] * 10 + sys.version_info[1]
      if ver >= 34:
          self._encodings.add("pickle-v4")
          self._encodings.add("gzip+pickle-v4")
  
      self._is_done_set = False
  
      # If the mode is 'write', we initialize things to empty
      if self._mode == "w":
          self._objects = {}
          self._info = {}
      elif self._mode == "r":
          if data_metadata is not None:
              # We already loaded the data metadata so just use that
              self._objects = data_metadata.get("objects", {})
              self._info = data_metadata.get("info", {})
          else:
              # What is the latest attempt ID for this task store.
              # NOTE: We *only* access to the data if the attempt that
              # produced it is done. In particular, we do not allow access to
              # a past attempt if a new attempt has started to avoid
              # inconsistencies (depending on when the user accesses the
              # datastore, the data may change). We make an exception to that
              # rule when allow_not_done is True which allows access to things
              # like logs even for tasks that did not write a done marker
              max_attempt = None
              for i in range(metaflow_config.MAX_ATTEMPTS):
                  check_meta = self._metadata_name_for_attempt(
                      self.METADATA_ATTEMPT_SUFFIX, i
                  )
                  if self.has_metadata(check_meta, add_attempt=False):
                      max_attempt = i
              if self._attempt is None:
                  self._attempt = max_attempt
              elif max_attempt is None or self._attempt > max_attempt:
                  # In this case the attempt does not exist, so we can't load
                  # anything
                  self._objects = {}
                  self._info = {}
                  return
  
              # Check if the latest attempt was completed successfully except
              # if we have allow_not_done
              data_obj = None
              if self.has_metadata(self.METADATA_DONE_SUFFIX):
                  data_obj = self.load_metadata(
                      [self.METADATA_DATA_SUFFIX]
                  )
                  data_obj = data_obj[self.METADATA_DATA_SUFFIX]
              elif self._attempt is None or not allow_not_done:
                  raise DataException(
                      "No completed attempts of the task was found for task '%s'"
                      % self._path
                  )
  
              if data_obj is not None:
                  self._objects = data_obj.get("objects", {})
                  self._info = data_obj.get("info", {})
      else:
          raise DataException(
              "Unknown datastore mode: '%s'" % self._mode
          )
  
imports: from collections import defaultdict
  import json
  import pickle
  import sys
  import time
  from functools import wraps
  from io import BufferedIOBase, FileIO, RawIOBase
  from types import MethodType, FunctionType
  from ..None import metaflow_config
  from ..exception import MetaflowInternalError
  from ..metadata import DataArtifact, MetaDatum
  from ..parameters import Parameter
  from ..util import Path, is_stringish, to_fileobj
  from .exceptions import DataException, UnpicklableArtifactException
label_desc: Use image node_id 3 to create a new TaskDataStore object from inherited base classes: object with example: obj = TaskDataStore(flow_datastore, run_id, step_name, task_id, attempt, data_metadata, mode, allow_not_done)