class_docstr: Shared context that holds all paths and storage utilities, passed along from
  the driver to workers.
  
  The properties of this context may not all be set at once, depending on where
  the context lives.
  For example, on the driver, the storage context is initialized, only knowing
  the experiment path. On the Trainable actor, the trial_dir_name is accessible.
  
  There are 2 types of paths:
  1. *_fs_path: A path on the `storage_filesystem`. This is a regular path
      which has been prefix-stripped by pyarrow.fs.FileSystem.from_uri and
      can be joined with `os.path.join`.
  2. *_local_path: The path on the local filesystem where results are saved to
     before persisting to storage.
  
  Example with storage_path="mock:///bucket/path?param=1":
  
      >>> from ray.train._internal.storage import StorageContext
      >>> import os
      >>> os.environ["RAY_AIR_LOCAL_CACHE_DIR"] = "/tmp/ray_results"
      >>> storage = StorageContext(
      ...     storage_path="mock://netloc/bucket/path?param=1",
      ...     experiment_dir_name="exp_name",
      ... )
      >>> storage.storage_filesystem   # Auto-resolved  # doctest: +ELLIPSIS
      <pyarrow._fs._MockFileSystem object...
      >>> storage.experiment_fs_path
      'bucket/path/exp_name'
      >>> storage.experiment_local_path
      '/tmp/ray_results/exp_name'
      >>> storage.trial_dir_name = "trial_dir"
      >>> storage.trial_fs_path
      'bucket/path/exp_name/trial_dir'
      >>> storage.trial_local_path
      '/tmp/ray_results/exp_name/trial_dir'
      >>> storage.current_checkpoint_index = 1
      >>> storage.checkpoint_fs_path
      'bucket/path/exp_name/trial_dir/checkpoint_000001'
  
  Example with storage_path=None:
  
      >>> from ray.train._internal.storage import StorageContext
      >>> import os
      >>> os.environ["RAY_AIR_LOCAL_CACHE_DIR"] = "/tmp/ray_results"
      >>> storage = StorageContext(
      ...     storage_path=None,
      ...     experiment_dir_name="exp_name",
      ... )
      >>> storage.storage_fs_path  # Auto-resolved
      '/tmp/ray_results'
      >>> storage.storage_local_path
      '/tmp/ray_results'
      >>> storage.experiment_local_path
      '/tmp/ray_results/exp_name'
      >>> storage.experiment_fs_path
      '/tmp/ray_results/exp_name'
      >>> storage.syncer is None
      True
      >>> storage.storage_filesystem   # Auto-resolved  # doctest: +ELLIPSIS
      <pyarrow._fs.LocalFileSystem object...
  
  Internal Usage Examples:
  - To copy files to the trial directory on the storage filesystem:
  
      pyarrow.fs.copy_files(
          local_dir,
          os.path.join(storage.trial_fs_path, "subdir"),
          destination_filesystem=storage.filesystem
      )
class_docstr_tok: ["Shared","context","that","holds","all","paths","and","storage","utilities",",","passed","along","from","the","driver","to","workers",".","The","properties","of","this","context","may","not","all","be","set","at","once",",","depending","on","where","the","context","lives",".","For","example",",","on","the","driver",",","the","storage","context","is","initialized",",","only","knowing","the","experiment","path",".","On","the","Trainable","actor",",","the","trial_dir_name","is","accessible",".","There","are","2","types","of","paths",":","1",".","*","_fs_path",":","A","path","on","the","`","storage_filesystem","`",".","This","is","a","regular","path","which","has","been","prefix-stripped","by","pyarrow.fs.FileSystem.from_uri","and","can","be","joined","with","`","os.path.join","`",".","2",".","*","_local_path",":","The","path","on","the","local","filesystem","where","results","are","saved","to","before","persisting","to","storage",".","Example","with","storage_path=","''","mock",":","\/\/\/bucket\/path","?","param=1","''",":",">",">",">","from","ray.train._internal.storage","import","StorageContext",">",">",">","import","os",">",">",">","os.environ","[","``","RAY_AIR_LOCAL_CACHE_DIR","''","]","=","``","\/tmp\/ray_results","''",">",">",">","storage","=","StorageContext","(","...","storage_path=","''","mock",":","\/\/netloc\/bucket\/path","?","param=1","''",",","...","experiment_dir_name=","''","exp_name","''",",","...",")",">",">",">","storage.storage_filesystem","#","Auto-resolved","#","doctest",":","+ELLIPSIS","<","pyarrow._fs._MockFileSystem","object","...",">",">",">","storage.experiment_fs_path","'bucket\/path\/exp_name'",">",">",">","storage.experiment_local_path","'\/tmp\/ray_results\/exp_name'",">",">",">","storage.trial_dir_name","=","``","trial_dir","''",">",">",">","storage.trial_fs_path","'bucket\/path\/exp_name\/trial_dir'",">",">",">","storage.trial_local_path","'\/tmp\/ray_results\/exp_name\/trial_dir'",">",">",">","storage.current_checkpoint_index","=","1",">",">",">","storage.checkpoint_fs_path","'bucket\/path\/exp_name\/trial_dir\/checkpoint_000001'","Example","with","storage_path=None",":",">",">",">","from","ray.train._internal.storage","import","StorageContext",">",">",">","import","os",">",">",">","os.environ","[","``","RAY_AIR_LOCAL_CACHE_DIR","''","]","=","``","\/tmp\/ray_results","''",">",">",">","storage","=","StorageContext","(","...","storage_path=None",",","...","experiment_dir_name=","''","exp_name","''",",","...",")",">",">",">","storage.storage_fs_path","#","Auto-resolved","'\/tmp\/ray_results'",">",">",">","storage.storage_local_path","'\/tmp\/ray_results'",">",">",">","storage.experiment_local_path","'\/tmp\/ray_results\/exp_name'",">",">",">","storage.experiment_fs_path","'\/tmp\/ray_results\/exp_name'",">",">",">","storage.syncer","is","None","True",">",">",">","storage.storage_filesystem","#","Auto-resolved","#","doctest",":","+ELLIPSIS","<","pyarrow._fs.LocalFileSystem","object","...","Internal","Usage","Examples",":","-","To","copy","files","to","the","trial","directory","on","the","storage","filesystem",":","pyarrow.fs.copy_files","(","local_dir",",","os.path.join","(","storage.trial_fs_path",",","``","subdir","''",")",",","destination_filesystem=storage.filesystem",")"]
docstr: Persists a given checkpoint to the current checkpoint path on the filesystem.
  
  "Current" is defined by the `current_checkpoint_index` attribute of the
  storage context.
  
  This method copies the checkpoint files to the storage location.
  It's up to the user to delete the original checkpoint files if desired.
  
  For example, the original directory is typically a local temp directory.
  
  Args:
      checkpoint: The checkpoint to persist to (fs, checkpoint_fs_path).
  
  Returns:
      Checkpoint: A Checkpoint pointing to the persisted checkpoint location.
docstr_tok: ["Persists","a","given","checkpoint","to","the","current","checkpoint","path","on","the","filesystem",".","``","Current","''","is","defined","by","the","`","current_checkpoint_index","`","attribute","of","the","storage","context",".","This","method","copies","the","checkpoint","files","to","the","storage","location",".","It","'s","up","to","the","user","to","delete","the","original","checkpoint","files","if","desired",".","For","example",",","the","original","directory","is","typically","a","local","temp","directory",".","Args",":","checkpoint",":","The","checkpoint","to","persist","to","(","fs",",","checkpoint_fs_path",")",".","Returns",":","Checkpoint",":","A","Checkpoint","pointing","to","the","persisted","checkpoint","location","."]
code: def persist_current_checkpoint(
      self, checkpoint: "Checkpoint"
  ) -> "Checkpoint":
      """Persists a given checkpoint to the current checkpoint path on the filesystem.
  
      "Current" is defined by the `current_checkpoint_index` attribute of the
      storage context.
  
      This method copies the checkpoint files to the storage location.
      It's up to the user to delete the original checkpoint files if desired.
  
      For example, the original directory is typically a local temp directory.
  
      Args:
          checkpoint: The checkpoint to persist to (fs, checkpoint_fs_path).
  
      Returns:
          Checkpoint: A Checkpoint pointing to the persisted checkpoint location.
      """
      # TODO(justinvyu): Fix this cyclical import.
      from ray.train._checkpoint import Checkpoint
  
      logger.debug(
          "Copying checkpoint files to storage path:\n"
          "({source_fs}, {source}) -> ({dest_fs}, {destination})".format(
              source=checkpoint.path,
              destination=self.checkpoint_fs_path,
              source_fs=checkpoint.filesystem,
              dest_fs=self.storage_filesystem,
          )
      )
  
      # Raise an error if the storage path is not accessible when
      # attempting to upload a checkpoint from a remote worker.
      # Ex: If storage_path is a local path, then a validation marker
      # will only exist on the head node but not the worker nodes.
      self._check_validation_file()
  
      self.storage_filesystem.create_dir(self.checkpoint_fs_path)
      _pyarrow_fs_copy_files(
          source=checkpoint.path,
          destination=self.checkpoint_fs_path,
          source_filesystem=checkpoint.filesystem,
          destination_filesystem=self.storage_filesystem,
      )
  
      persisted_checkpoint = Checkpoint(
          filesystem=self.storage_filesystem,
          path=self.checkpoint_fs_path,
      )
      logger.info(
          f"Checkpoint successfully created at: {persisted_checkpoint}"
      )
      return persisted_checkpoint
  
imports: import dataclasses
  import fnmatch
  import logging
  import os
  import shutil
  from pathlib import Path
  from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Tuple, Type, Union
  from ray._private.storage import _get_storage_uri
  from ray.air._internal.filelock import TempFileLock
  from ray.train._internal.syncer import SyncConfig, Syncer, _BackgroundSyncer
  from ray.train.constants import _get_defaults_results_dir
label_desc: Use image node_id 6 for calling the StorageContext obj's underlying member method code with example usage: obj.persist_current_checkpoint(checkpoint) and returns: persisted_checkpoint