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",")","."]
docstr: Persist any new artifacts that were produced when running flow
  
  NOTE: This is a DESTRUCTIVE operation that deletes artifacts from
  the given flow to conserve memory. Don't rely on artifact attributes
  of the flow object after calling this function.
  
  Parameters
  ----------
  flow : FlowSpec
      Flow to persist
docstr_tok: ["Persist","any","new","artifacts","that","were","produced","when","running","flow","NOTE",":","This","is","a","DESTRUCTIVE","operation","that","deletes","artifacts","from","the","given","flow","to","conserve","memory",".","Do","n't","rely","on","artifact","attributes","of","the","flow","object","after","calling","this","function",".","Parameters","--","--","--","--","--","flow",":","FlowSpec","Flow","to","persist"]
code: def persist(self, flow):
      """
      Persist any new artifacts that were produced when running flow
  
      NOTE: This is a DESTRUCTIVE operation that deletes artifacts from
      the given flow to conserve memory. Don't rely on artifact attributes
      of the flow object after calling this function.
  
      Parameters
      ----------
      flow : FlowSpec
          Flow to persist
      """
  
      if flow._datastore:
          self._objects.update(flow._datastore._objects)
          self._info.update(flow._datastore._info)
  
      # we create a list of valid_artifacts in advance, outside of
      # artifacts_iter, so we can provide a len_hint below
      valid_artifacts = []
      for var in dir(flow):
          if var.startswith("__") or var in flow._EPHEMERAL:
              continue
          # Skip over properties of the class (Parameters or class variables)
          if hasattr(flow.__class__, var) and isinstance(
              getattr(flow.__class__, var), property
          ):
              continue
  
          val = getattr(flow, var)
          if not (
              isinstance(val, MethodType)
              or isinstance(val, FunctionType)
              or isinstance(val, Parameter)
          ):
              valid_artifacts.append((var, val))
  
      def artifacts_iter():
          # we consume the valid_artifacts list destructively to
          # make sure we don't keep references to artifacts. We
          # want to avoid keeping original artifacts and encoded
          # artifacts in memory simultaneously
          while valid_artifacts:
              var, val = valid_artifacts.pop()
              if not var.startswith("_") and var != "name":
                  # NOTE: Destructive mutation of the flow object. We keep
                  # around artifacts called 'name' and anything starting with
                  # '_' as they are used by the Metaflow runtime.
                  delattr(flow, var)
              yield var, val
  
      self.save_artifacts(
          artifacts_iter(), len_hint=len(valid_artifacts)
      )
  
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 29 for calling the TaskDataStore obj's underlying member method code with example usage: obj.persist(flow) without return types