class_docstr: Store and load checkpoints on GCS. Although GCS is similar to S3, some
  S3 APIs are not supported on GCS and vice versa. Moreover, Google
  recommends using the google-storage-python library to access GCS,
  rather than the boto library we use to access S3 -- boto uses
  various S3 features that are not supported by GCS.
  
  Batching is supported by the GCS API for deletion, however it is not used because
  of observed request failures. Batching is not used for uploading
  or downloading files, because the GCS API does not support it. Upload/download
  performance could be improved by using multiple clients in a multithreaded fashion.
  
  Authentication is currently only supported via the "Application
  Default Credentials" method in GCP [1]. Typical configuration:
  ensure your VM runs in a service account that has sufficient
  permissions to read/write/delete from the GCS bucket where
  checkpoints will be stored (this only works when running in GCE).
code: def download(
      self,
      src: str,
      dst: Union[str, os.PathLike],
      selector: Optional[storage.Selector] = None,
  ) -> None:
      from google.api_core import exceptions as api_exceptions
      from google.auth import exceptions as auth_exceptions
  
      dst = os.fspath(dst)
      path = self.get_storage_prefix(src)
      logger.info(f"Downloading {path} from GCS")
      found = False
  
      # Listing blobs with prefix set and no delimiter is equivalent to a recursive listing.  If
      # you include a `delimiter="/"` you will get only the file-like blobs inside of a
      # directory-like blob.
      try:
          for blob in self.bucket.list_blobs(prefix=path):
              found = True
              relname = os.path.relpath(blob.name, path)
              if blob.name.endswith("/"):
                  relname = os.path.join(relname, "")
              if selector is not None and not selector(relname):
                  continue
              _dst = os.path.join(dst, relname)
              dst_dir = os.path.dirname(_dst)
              if not os.path.exists(dst_dir):
                  os.makedirs(dst_dir, exist_ok=True)
  
              # Only create empty directory for keys that end with "/".
              # See `upload` method for more context.
              if blob.name.endswith("/"):
                  os.makedirs(_dst, exist_ok=True)
                  continue
  
              logger.debug(f"Downloading from GCS: {blob.name}")
  
              blob.download_to_filename(_dst)
  
      except (
          auth_exceptions.GoogleAuthError,
          api_exceptions.Unauthorized,
          api_exceptions.Forbidden,
      ) as e:
          raise errors.NoDirectStorageAccess(
              "Unable to access cloud checkpoint storage"
          ) from e
  
      if not found:
          raise errors.CheckpointNotFound(
              f"Did not find checkpoint {path} in GCS"
          )
  
file_path: determined/harness/determined/common/storage/gcs.py
imports: import logging
  import os
  import tempfile
  from typing import Dict, List, Optional, Union, no_type_check
  import requests.exceptions
  import urllib3.exceptions
  from determined import errors
  from determined.common import storage, util
  from determined.common.storage.s3 import normalize_prefix
label_desc: Use image node_id 4 for calling the GCSStorageManager obj's underlying member method code with example usage: obj.download(src, dst, selector) without return types