class_docstr: Abstract artifact repo that defines how to upload (log) and download potentially large
  artifacts from different storage backends.
class_docstr_tok: ["Abstract","artifact","repo","that","defines","how","to","upload","(","log",")","and","download","potentially","large","artifacts","from","different","storage","backends","."]
docstr: Download an artifact file or directory to a local directory if applicable, and return a
  local path for it.
  The caller is responsible for managing the lifecycle of the downloaded artifacts.
  
  :param artifact_path: Relative source path to the desired artifacts.
  :param dst_path: Absolute path of the local filesystem destination directory to which to
                   download the specified artifacts. This directory must already exist.
                   If unspecified, the artifacts will either be downloaded to a new
                   uniquely-named directory on the local filesystem or will be returned
                   directly in the case of the LocalArtifactRepository.
  
  :return: Absolute path of the local filesystem location containing the desired artifacts.
docstr_tok: ["Download","an","artifact","file","or","directory","to","a","local","directory","if","applicable",",","and","return","a","local","path","for","it",".","The","caller","is","responsible","for","managing","the","lifecycle","of","the","downloaded","artifacts",".",":","param","artifact_path",":","Relative","source","path","to","the","desired","artifacts",".",":","param","dst_path",":","Absolute","path","of","the","local","filesystem","destination","directory","to","which","to","download","the","specified","artifacts",".","This","directory","must","already","exist",".","If","unspecified",",","the","artifacts","will","either","be","downloaded","to","a","new","uniquely-named","directory","on","the","local","filesystem","or","will","be","returned","directly","in","the","case","of","the","LocalArtifactRepository",".",":","return",":","Absolute","path","of","the","local","filesystem","location","containing","the","desired","artifacts","."]
code: def download_artifacts(self, artifact_path, dst_path=None):
      """
      Download an artifact file or directory to a local directory if applicable, and return a
      local path for it.
      The caller is responsible for managing the lifecycle of the downloaded artifacts.
  
      :param artifact_path: Relative source path to the desired artifacts.
      :param dst_path: Absolute path of the local filesystem destination directory to which to
                       download the specified artifacts. This directory must already exist.
                       If unspecified, the artifacts will either be downloaded to a new
                       uniquely-named directory on the local filesystem or will be returned
                       directly in the case of the LocalArtifactRepository.
  
      :return: Absolute path of the local filesystem location containing the desired artifacts.
      """
      if dst_path:
          dst_path = os.path.abspath(dst_path)
          if not os.path.exists(dst_path):
              raise MlflowException(
                  message=(
                      "The destination path for downloaded artifacts does not"
                      f" exist! Destination path: {dst_path}"
                  ),
                  error_code=RESOURCE_DOES_NOT_EXIST,
              )
          elif not os.path.isdir(dst_path):
              raise MlflowException(
                  message=(
                      "The destination path for downloaded artifacts must be a directory!"
                      f" Destination path: {dst_path}"
                  ),
                  error_code=INVALID_PARAMETER_VALUE,
              )
      else:
          dst_path = create_tmp_dir()
  
      def _download_file(src_artifact_path, dst_local_dir_path):
          dst_local_file_path = self._create_download_destination(
              src_artifact_path=src_artifact_path,
              dst_local_dir_path=dst_local_dir_path,
          )
          return self.thread_pool.submit(
              self._download_file,
              remote_file_path=src_artifact_path,
              local_path=dst_local_file_path,
          )
  
      # Submit download tasks
      futures = {}
      if self._is_directory(artifact_path):
          for file_info in self._iter_artifacts_recursive(
              artifact_path
          ):
              if file_info.is_dir:  # Empty directory
                  os.makedirs(
                      os.path.join(dst_path, file_info.path),
                      exist_ok=True,
                  )
              else:
                  fut = _download_file(file_info.path, dst_path)
                  futures[fut] = file_info.path
      else:
          fut = _download_file(artifact_path, dst_path)
          futures[fut] = artifact_path
  
      # Wait for downloads to complete and collect failures
      failed_downloads = {}
      with ArtifactProgressBar.files(
          desc="Downloading artifacts", total=len(futures)
      ) as pbar:
          if len(futures) >= 10 and pbar.pbar:
              _logger.info(
                  "The progress bar can be disabled by setting the environment "
                  f"variable {MLFLOW_ENABLE_ARTIFACTS_PROGRESS_BAR} to false"
              )
          for f in as_completed(futures):
              try:
                  f.result()
                  pbar.update()
              except Exception as e:
                  path = futures[f]
                  failed_downloads[path] = e
  
      if failed_downloads:
          template = "##### File {path} #####\n{error}"
          failures = "\n".join(
              template.format(path=path, error=error)
              for path, error in failed_downloads.items()
          )
          raise MlflowException(
              message=(
                  "The following failures occurred while downloading one or more"
                  f" artifacts from {self.artifact_uri}:\n{_truncate_error(failures)}"
              )
          )
  
      return os.path.join(dst_path, artifact_path)
  
imports: import logging
  import os
  import posixpath
  from abc import ABC, ABCMeta, abstractmethod
  from concurrent.futures import ThreadPoolExecutor, as_completed
  from typing import List, Optional
  from mlflow.entities.file_info import FileInfo
  from mlflow.entities.multipart_upload import CreateMultipartUploadResponse, MultipartUploadPart
  from mlflow.environment_variables import MLFLOW_ENABLE_ARTIFACTS_PROGRESS_BAR
  from mlflow.exceptions import MlflowException
  from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE, RESOURCE_DOES_NOT_EXIST
  from mlflow.utils.annotations import developer_stable
  from mlflow.utils.file_utils import ArtifactProgressBar, create_tmp_dir
  from mlflow.utils.validation import bad_path_message, path_not_unique
label_desc: Use image node_id 9 for calling the ArtifactRepository obj's underlying member method code with example usage: obj.download_artifacts(artifact_path, dst_path) and returns: os, self