class_docstr: Data provider for managing experiment lifecycle.
class_docstr_tok: ["Data","provider","for","managing","experiment","lifecycle","."]
code: def checkpoint_experiment(self, experiment):
      key = self._experiment_key(experiment)
      key_path = self._get_experiments_keybase() + key
  
      self.logger.debug("checkpointing %s", key_path)
  
      experiment_dict = self._get(key_path)
  
      workers = []
      with ThreadPoolExecutor(max_workers=8) as executor:
          for _, art in experiment.artifacts.items():
              if art.is_mutable and art.local_path is not None:
                  workers.append(executor.submit(art.upload, None))
          wait(workers)
  
      for worker in workers:
          try:
              worker.result()
          except Exception as exc:
              # If any of artifact savers failed and threw an exception,
              # rethrow it to signal overall checkpoint failure
              raise exc
  
      checkpoint_time = time.time()
      experiment.time_last_checkpoint = checkpoint_time
      experiment_dict["time_last_checkpoint"] = checkpoint_time
  
      self._set(key_path, experiment_dict)
  
imports: import time
  from concurrent.futures import ThreadPoolExecutor, wait
  from studio.util import util, logs
  from studio.storage.storage_handler import StorageHandler
  from studio.artifacts.artifact import Artifact
  from studio.experiments.experiment import Experiment, experiment_from_dict
  from studio.storage.storage_setup import get_storage_verbose_level
  from studio.util.util import retry, report_fatal, compression_to_extension, check_for_kb_interrupt
label_desc: Use image node_id 17 for calling the KeyValueProvider obj's underlying member method code with example usage: obj.checkpoint_experiment(experiment) without return types