code: def _download_from_fs_path(
      fs: pyarrow.fs.FileSystem,
      fs_path: str,
      local_path: str,
      filelock: bool = True,
  ):
      """Downloads a directory or file from (fs, fs_path) to a local path.
  
      If fs_path points to a directory:
      - The full directory contents are downloaded directly into `local_path`,
        rather than to a subdirectory of `local_path`.
  
      If fs_path points to a file:
      - The file is downloaded to `local_path`, which is expected to be a file path.
  
      If the download fails, the `local_path` contents are
      cleaned up before raising, if the directory did not previously exist.
  
      NOTE: This method creates `local_path`'s parent directories if they do not
      already exist. If the download fails, this does NOT clean up all the parent
      directories that were created.
  
      Args:
          fs: The filesystem to download from.
          fs_path: The filesystem path (either a directory or a file) to download.
          local_path: The local path to download to.
          filelock: Whether to require a file lock before downloading, useful for
              multiple downloads to the same directory that may be happening in parallel.
  
      Raises:
          FileNotFoundError: if (fs, fs_path) doesn't exist.
      """
  
      _local_path = Path(local_path).resolve()
      exists_before = _local_path.exists()
      if _is_directory(fs=fs, fs_path=fs_path):
          _local_path.mkdir(parents=True, exist_ok=True)
      else:
          _local_path.parent.mkdir(parents=True, exist_ok=True)
  
      try:
          if filelock:
              with TempFileLock(f"{os.path.normpath(local_path)}.lock"):
                  _pyarrow_fs_copy_files(
                      fs_path, local_path, source_filesystem=fs
                  )
          else:
              _pyarrow_fs_copy_files(
                  fs_path, local_path, source_filesystem=fs
              )
      except Exception as e:
          # Clean up the directory if downloading was unsuccessful
          if not exists_before:
              shutil.rmtree(local_path, ignore_errors=True)
          raise e
  
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 3 for calling a global function with example usage: _download_from_fs_path(fs, fs_path, local_path, filelock) without return types