Aluode/PerceptionLabPortable
0
1from __future__ import annotations2 3import io4import json5import logging6import os7import threading8import warnings9import weakref10from errno import ESPIPE11from glob import has_magic12from hashlib import sha25613from typing import Any, ClassVar14 15from .callbacks import DEFAULT_CALLBACK16from .config import apply_config, conf17from .dircache import DirCache18from .transaction import Transaction19from .utils import (20 _unstrip_protocol,21 glob_translate,22 isfilelike,23 other_paths,24 read_block,25 stringify_path,26 tokenize,27)28 29logger = logging.getLogger("fsspec")30 31 32def make_instance(cls, args, kwargs):33 return cls(*args, **kwargs)34 35 36class _Cached(type):37 """38 Metaclass for caching file system instances.39 40 Notes41 -----42 Instances are cached according to43 44 * The values of the class attributes listed in `_extra_tokenize_attributes`45 * The arguments passed to ``__init__``.46 47 This creates an additional reference to the filesystem, which prevents the48 filesystem from being garbage collected when all *user* references go away.49 A call to the :meth:`AbstractFileSystem.clear_instance_cache` must *also*50 be made for a filesystem instance to be garbage collected.51 """52 53 def __init__(cls, *args, **kwargs):54 super().__init__(*args, **kwargs)55 # Note: we intentionally create a reference here, to avoid garbage56 # collecting instances when all other references are gone. To really57 # delete a FileSystem, the cache must be cleared.58 if conf.get("weakref_instance_cache"): # pragma: no cover59 # debug option for analysing fork/spawn conditions60 cls._cache = weakref.WeakValueDictionary()61 else:62 cls._cache = {}63 cls._pid = os.getpid()64 65 def __call__(cls, *args, **kwargs):66 kwargs = apply_config(cls, kwargs)67 extra_tokens = tuple(68 getattr(cls, attr, None) for attr in cls._extra_tokenize_attributes69 )70 strip_tokenize_options = {71 k: kwargs.pop(k) for k in cls._strip_tokenize_options if k in kwargs72 }73 token = tokenize(74 cls, cls._pid, threading.get_ident(), *args, *extra_tokens, **kwargs75 )76 skip = kwargs.pop("skip_instance_cache", False)77 if os.getpid() != cls._pid:78 cls._cache.clear()79 cls._pid = os.getpid()80 if not skip and cls.cachable and token in cls._cache:81 cls._latest = token82 return cls._cache[token]83 else:84 obj = super().__call__(*args, **kwargs, **strip_tokenize_options)85 # Setting _fs_token here causes some static linters to complain.86 obj._fs_token_ = token87 obj.storage_args = args88 obj.storage_options = kwargs89 if obj.async_impl and obj.mirror_sync_methods:90 from .asyn import mirror_sync_methods91 92 mirror_sync_methods(obj)93 94 if cls.cachable and not skip:95 cls._latest = token96 cls._cache[token] = obj97 return obj98 99 100class AbstractFileSystem(metaclass=_Cached):101 """102 An abstract super-class for pythonic file-systems103 104 Implementations are expected to be compatible with or, better, subclass105 from here.106 """107 108 cachable = True # this class can be cached, instances reused109 _cached = False110 blocksize = 2**22111 sep = "/"112 protocol: ClassVar[str | tuple[str, ...]] = "abstract"113 _latest = None114 async_impl = False115 mirror_sync_methods = False116 root_marker = "" # For some FSs, may require leading '/' or other character117 transaction_type = Transaction118 119 #: Extra *class attributes* that should be considered when hashing.120 _extra_tokenize_attributes = ()121 #: *storage options* that should not be considered when hashing.122 _strip_tokenize_options = ()123 124 # Set by _Cached metaclass125 storage_args: tuple[Any, ...]126 storage_options: dict[str, Any]127 128 def __init__(self, *args, **storage_options):129 """Create and configure file-system instance130 131 Instances may be cachable, so if similar enough arguments are seen132 a new instance is not required. The token attribute exists to allow133 implementations to cache instances if they wish.134 135 A reasonable default should be provided if there are no arguments.136 137 Subclasses should call this method.138 139 Parameters140 ----------141 use_listings_cache, listings_expiry_time, max_paths:142 passed to ``DirCache``, if the implementation supports143 directory listing caching. Pass use_listings_cache=False144 to disable such caching.145 skip_instance_cache: bool146 If this is a cachable implementation, pass True here to force147 creating a new instance even if a matching instance exists, and prevent148 storing this instance.149 asynchronous: bool150 loop: asyncio-compatible IOLoop or None151 """152 if self._cached:153 # reusing instance, don't change154 return155 self._cached = True156 self._intrans = False157 self._transaction = None158 self._invalidated_caches_in_transaction = []159 self.dircache = DirCache(**storage_options)160 161 if storage_options.pop("add_docs", None):162 warnings.warn("add_docs is no longer supported.", FutureWarning)163 164 if storage_options.pop("add_aliases", None):165 warnings.warn("add_aliases has been removed.", FutureWarning)166 # This is set in _Cached167 self._fs_token_ = None168 169 @property170 def fsid(self):171 """Persistent filesystem id that can be used to compare filesystems172 across sessions.173 """174 raise NotImplementedError175 176 @property177 def _fs_token(self):178 return self._fs_token_179 180 def __dask_tokenize__(self):181 return self._fs_token182 183 def __hash__(self):184 return int(self._fs_token, 16)185 186 def __eq__(self, other):187 return isinstance(other, type(self)) and self._fs_token == other._fs_token188 189 def __reduce__(self):190 return make_instance, (type(self), self.storage_args, self.storage_options)191 192 @classmethod193 def _strip_protocol(cls, path):194 """Turn path from fully-qualified to file-system-specific195 196 May require FS-specific handling, e.g., for relative paths or links.197 """198 if isinstance(path, list):199 return [cls._strip_protocol(p) for p in path]200 path = stringify_path(path)201 protos = (cls.protocol,) if isinstance(cls.protocol, str) else cls.protocol202 for protocol in protos:203 if path.startswith(protocol + "://"):204 path = path[len(protocol) + 3 :]205 elif path.startswith(protocol + "::"):206 path = path[len(protocol) + 2 :]207 path = path.rstrip("/")208 # use of root_marker to make minimum required path, e.g., "/"209 return path or cls.root_marker210 211 def unstrip_protocol(self, name: str) -> str:212 """Format FS-specific path to generic, including protocol"""213 protos = (self.protocol,) if isinstance(self.protocol, str) else self.protocol214 for protocol in protos:215 if name.startswith(f"{protocol}://"):216 return name217 return f"{protos[0]}://{name}"218 219 @staticmethod220 def _get_kwargs_from_urls(path):221 """If kwargs can be encoded in the paths, extract them here222 223 This should happen before instantiation of the class; incoming paths224 then should be amended to strip the options in methods.225 226 Examples may look like an sftp path "sftp://user@host:/my/path", where227 the user and host should become kwargs and later get stripped.228 """229 # by default, nothing happens230 return {}231 232 @classmethod233 def current(cls):234 """Return the most recently instantiated FileSystem235 236 If no instance has been created, then create one with defaults237 """238 if cls._latest in cls._cache:239 return cls._cache[cls._latest]240 return cls()241 242 @property243 def transaction(self):244 """A context within which files are committed together upon exit245 246 Requires the file class to implement `.commit()` and `.discard()`247 for the normal and exception cases.248 """249 if self._transaction is None:250 self._transaction = self.transaction_type(self)251 return self._transaction252 253 def start_transaction(self):254 """Begin write transaction for deferring files, non-context version"""255 self._intrans = True256 self._transaction = self.transaction_type(self)257 return self.transaction258 259 def end_transaction(self):260 """Finish write transaction, non-context version"""261 self.transaction.complete()262 self._transaction = None263 # The invalid cache must be cleared after the transaction is completed.264 for path in self._invalidated_caches_in_transaction:265 self.invalidate_cache(path)266 self._invalidated_caches_in_transaction.clear()267 268 def invalidate_cache(self, path=None):269 """270 Discard any cached directory information271 272 Parameters273 ----------274 path: string or None275 If None, clear all listings cached else listings at or under given276 path.277 """278 # Not necessary to implement invalidation mechanism, may have no cache.279 # But if have, you should call this method of parent class from your280 # subclass to ensure expiring caches after transacations correctly.281 # See the implementation of FTPFileSystem in ftp.py282 if self._intrans:283 self._invalidated_caches_in_transaction.append(path)284 285 def mkdir(self, path, create_parents=True, **kwargs):286 """287 Create directory entry at path288 289 For systems that don't have true directories, may create an for290 this instance only and not touch the real filesystem291 292 Parameters293 ----------294 path: str295 location296 create_parents: bool297 if True, this is equivalent to ``makedirs``298 kwargs:299 may be permissions, etc.300 """301 pass # not necessary to implement, may not have directories302 303 def makedirs(self, path, exist_ok=False):304 """Recursively make directories305 306 Creates directory at path and any intervening required directories.307 Raises exception if, for instance, the path already exists but is a308 file.309 310 Parameters311 ----------312 path: str313 leaf directory name314 exist_ok: bool (False)315 If False, will error if the target already exists316 """317 pass # not necessary to implement, may not have directories318 319 def rmdir(self, path):320 """Remove a directory, if empty"""321 pass # not necessary to implement, may not have directories322 323 def ls(self, path, detail=True, **kwargs):324 """List objects at path.325 326 This should include subdirectories and files at that location. The327 difference between a file and a directory must be clear when details328 are requested.329 330 The specific keys, or perhaps a FileInfo class, or similar, is TBD,331 but must be consistent across implementations.332 Must include:333 334 - full path to the entry (without protocol)335 - size of the entry, in bytes. If the value cannot be determined, will336 be ``None``.337 - type of entry, "file", "directory" or other338 339 Additional information340 may be present, appropriate to the file-system, e.g., generation,341 checksum, etc.342 343 May use refresh=True|False to allow use of self._ls_from_cache to344 check for a saved listing and avoid calling the backend. This would be345 common where listing may be expensive.346 347 Parameters348 ----------349 path: str350 detail: bool351 if True, gives a list of dictionaries, where each is the same as352 the result of ``info(path)``. If False, gives a list of paths353 (str).354 kwargs: may have additional backend-specific options, such as version355 information356 357 Returns358 -------359 List of strings if detail is False, or list of directory information360 dicts if detail is True.361 """362 raise NotImplementedError363 364 def _ls_from_cache(self, path):365 """Check cache for listing366 367 Returns listing, if found (may be empty list for a directly that exists368 but contains nothing), None if not in cache.369 """370 parent = self._parent(path)371 try:372 return self.dircache[path.rstrip("/")]373 except KeyError:374 pass375 try:376 files = [377 f378 for f in self.dircache[parent]379 if f["name"] == path380 or (f["name"] == path.rstrip("/") and f["type"] == "directory")381 ]382 if len(files) == 0:383 # parent dir was listed but did not contain this file384 raise FileNotFoundError(path)385 return files386 except KeyError:387 pass388 389 def walk(self, path, maxdepth=None, topdown=True, on_error="omit", **kwargs):390 """Return all files under the given path.391 392 List all files, recursing into subdirectories; output is iterator-style,393 like ``os.walk()``. For a simple list of files, ``find()`` is available.394 395 When topdown is True, the caller can modify the dirnames list in-place (perhaps396 using del or slice assignment), and walk() will397 only recurse into the subdirectories whose names remain in dirnames;398 this can be used to prune the search, impose a specific order of visiting,399 or even to inform walk() about directories the caller creates or renames before400 it resumes walk() again.401 Modifying dirnames when topdown is False has no effect. (see os.walk)402 403 Note that the "files" outputted will include anything that is not404 a directory, such as links.405 406 Parameters407 ----------408 path: str409 Root to recurse into410 maxdepth: int411 Maximum recursion depth. None means limitless, but not recommended412 on link-based file-systems.413 topdown: bool (True)414 Whether to walk the directory tree from the top downwards or from415 the bottom upwards.416 on_error: "omit", "raise", a callable417 if omit (default), path with exception will simply be empty;418 If raise, an underlying exception will be raised;419 if callable, it will be called with a single OSError instance as argument420 kwargs: passed to ``ls``421 """422 if maxdepth is not None and maxdepth < 1:423 raise ValueError("maxdepth must be at least 1")424 425 path = self._strip_protocol(path)426 full_dirs = {}427 dirs = {}428 files = {}429 430 detail = kwargs.pop("detail", False)431 try:432 listing = self.ls(path, detail=True, **kwargs)433 except (FileNotFoundError, OSError) as e:434 if on_error == "raise":435 raise436 if callable(on_error):437 on_error(e)438 return439 440 for info in listing:441 # each info name must be at least [path]/part , but here442 # we check also for names like [path]/part/443 pathname = info["name"].rstrip("/")444 name = pathname.rsplit("/", 1)[-1]445 if info["type"] == "directory" and pathname != path:446 # do not include "self" path447 full_dirs[name] = pathname448 dirs[name] = info449 elif pathname == path:450 # file-like with same name as give path451 files[""] = info452 else:453 files[name] = info454 455 if not detail:456 dirs = list(dirs)457 files = list(files)458 459 if topdown:460 # Yield before recursion if walking top down461 yield path, dirs, files462 463 if maxdepth is not None:464 maxdepth -= 1465 if maxdepth < 1:466 if not topdown:467 yield path, dirs, files468 return469 470 for d in dirs:471 yield from self.walk(472 full_dirs[d],473 maxdepth=maxdepth,474 detail=detail,475 topdown=topdown,476 **kwargs,477 )478 479 if not topdown:480 # Yield after recursion if walking bottom up481 yield path, dirs, files482 483 def find(self, path, maxdepth=None, withdirs=False, detail=False, **kwargs):484 """List all files below path.485 486 Like posix ``find`` command without conditions487 488 Parameters489 ----------490 path : str491 maxdepth: int or None492 If not None, the maximum number of levels to descend493 withdirs: bool494 Whether to include directory paths in the output. This is True495 when used by glob, but users usually only want files.496 kwargs are passed to ``ls``.497 """498 # TODO: allow equivalent of -name parameter499 path = self._strip_protocol(path)500 out = {}501 502 # Add the root directory if withdirs is requested503 # This is needed for posix glob compliance504 if withdirs and path != "" and self.isdir(path):505 out[path] = self.info(path)506 507 for _, dirs, files in self.walk(path, maxdepth, detail=True, **kwargs):508 if withdirs:509 files.update(dirs)510 out.update({info["name"]: info for name, info in files.items()})511 if not out and self.isfile(path):512 # walk works on directories, but find should also return [path]513 # when path happens to be a file514 out[path] = {}515 names = sorted(out)516 if not detail:517 return names518 else:519 return {name: out[name] for name in names}520 521 def du(self, path, total=True, maxdepth=None, withdirs=False, **kwargs):522 """Space used by files and optionally directories within a path523 524 Directory size does not include the size of its contents.525 526 Parameters527 ----------528 path: str529 total: bool530 Whether to sum all the file sizes531 maxdepth: int or None532 Maximum number of directory levels to descend, None for unlimited.533 withdirs: bool534 Whether to include directory paths in the output.535 kwargs: passed to ``find``536 537 Returns538 -------539 Dict of {path: size} if total=False, or int otherwise, where numbers540 refer to bytes used.541 """542 sizes = {}543 if withdirs and self.isdir(path):544 # Include top-level directory in output545 info = self.info(path)546 sizes[info["name"]] = info["size"]547 for f in self.find(path, maxdepth=maxdepth, withdirs=withdirs, **kwargs):548 info = self.info(f)549 sizes[info["name"]] = info["size"]550 if total:551 return sum(sizes.values())552 else:553 return sizes554 555 def glob(self, path, maxdepth=None, **kwargs):556 """Find files by glob-matching.557 558 Pattern matching capabilities for finding files that match the given pattern.559 560 Parameters561 ----------562 path: str563 The glob pattern to match against564 maxdepth: int or None565 Maximum depth for ``'**'`` patterns. Applied on the first ``'**'`` found.566 Must be at least 1 if provided.567 kwargs:568 Additional arguments passed to ``find`` (e.g., detail=True)569 570 Returns571 -------572 List of matched paths, or dict of paths and their info if detail=True573 574 Notes575 -----576 Supported patterns:577 - '*': Matches any sequence of characters within a single directory level578 - ``'**'``: Matches any number of directory levels (must be an entire path component)579 - '?': Matches exactly one character580 - '[abc]': Matches any character in the set581 - '[a-z]': Matches any character in the range582 - '[!abc]': Matches any character NOT in the set583 584 Special behaviors:585 - If the path ends with '/', only folders are returned586 - Consecutive '*' characters are compressed into a single '*'587 - Empty brackets '[]' never match anything588 - Negated empty brackets '[!]' match any single character589 - Special characters in character classes are escaped properly590 591 Limitations:592 - ``'**'`` must be a complete path component (e.g., ``'a/**/b'``, not ``'a**b'``)593 - No brace expansion ('{a,b}.txt')594 - No extended glob patterns ('+(pattern)', '!(pattern)')595 """596 if maxdepth is not None and maxdepth < 1:597 raise ValueError("maxdepth must be at least 1")598 599 import re600 601 seps = (os.path.sep, os.path.altsep) if os.path.altsep else (os.path.sep,)602 ends_with_sep = path.endswith(seps) # _strip_protocol strips trailing slash603 path = self._strip_protocol(path)604 append_slash_to_dirname = ends_with_sep or path.endswith(605 tuple(sep + "**" for sep in seps)606 )607 idx_star = path.find("*") if path.find("*") >= 0 else len(path)608 idx_qmark = path.find("?") if path.find("?") >= 0 else len(path)609 idx_brace = path.find("[") if path.find("[") >= 0 else len(path)610 611 min_idx = min(idx_star, idx_qmark, idx_brace)612 613 detail = kwargs.pop("detail", False)614 615 if not has_magic(path):616 if self.exists(path, **kwargs):617 if not detail:618 return [path]619 else:620 return {path: self.info(path, **kwargs)}621 else:622 if not detail:623 return [] # glob of non-existent returns empty624 else:625 return {}626 elif "/" in path[:min_idx]:627 min_idx = path[:min_idx].rindex("/")628 root = path[: min_idx + 1]629 depth = path[min_idx + 1 :].count("/") + 1630 else:631 root = ""632 depth = path[min_idx + 1 :].count("/") + 1633 634 if "**" in path:635 if maxdepth is not None:636 idx_double_stars = path.find("**")637 depth_double_stars = path[idx_double_stars:].count("/") + 1638 depth = depth - depth_double_stars + maxdepth639 else:640 depth = None641 642 allpaths = self.find(root, maxdepth=depth, withdirs=True, detail=True, **kwargs)643 644 pattern = glob_translate(path + ("/" if ends_with_sep else ""))645 pattern = re.compile(pattern)646 647 out = {648 p: info649 for p, info in sorted(allpaths.items())650 if pattern.match(651 p + "/"652 if append_slash_to_dirname and info["type"] == "directory"653 else p654 )655 }656 657 if detail:658 return out659 else:660 return list(out)661 662 def exists(self, path, **kwargs):663 """Is there a file at the given path"""664 try:665 self.info(path, **kwargs)666 return True667 except: # noqa: E722668 # any exception allowed bar FileNotFoundError?669 return False670 671 def lexists(self, path, **kwargs):672 """If there is a file at the given path (including673 broken links)"""674 return self.exists(path)675 676 def info(self, path, **kwargs):677 """Give details of entry at path678 679 Returns a single dictionary, with exactly the same information as ``ls``680 would with ``detail=True``.681 682 The default implementation calls ls and could be overridden by a683 shortcut. kwargs are passed on to ```ls()``.684 685 Some file systems might not be able to measure the file's size, in686 which case, the returned dict will include ``'size': None``.687 688 Returns689 -------690 dict with keys: name (full path in the FS), size (in bytes), type (file,691 directory, or something else) and other FS-specific keys.692 """693 path = self._strip_protocol(path)694 out = self.ls(self._parent(path), detail=True, **kwargs)695 out = [o for o in out if o["name"].rstrip("/") == path]696 if out:697 return out[0]698 out = self.ls(path, detail=True, **kwargs)699 path = path.rstrip("/")700 out1 = [o for o in out if o["name"].rstrip("/") == path]701 if len(out1) == 1:702 if "size" not in out1[0]:703 out1[0]["size"] = None704 return out1[0]705 elif len(out1) > 1 or out:706 return {"name": path, "size": 0, "type": "directory"}707 else:708 raise FileNotFoundError(path)709 710 def checksum(self, path):711 """Unique value for current version of file712 713 If the checksum is the same from one moment to another, the contents714 are guaranteed to be the same. If the checksum changes, the contents715 *might* have changed.716 717 This should normally be overridden; default will probably capture718 creation/modification timestamp (which would be good) or maybe719 access timestamp (which would be bad)720 """721 return int(tokenize(self.info(path)), 16)722 723 def size(self, path):724 """Size in bytes of file"""725 return self.info(path).get("size", None)726 727 def sizes(self, paths):728 """Size in bytes of each file in a list of paths"""729 return [self.size(p) for p in paths]730 731 def isdir(self, path):732 """Is this entry directory-like?"""733 try:734 return self.info(path)["type"] == "directory"735 except OSError:736 return False737 738 def isfile(self, path):739 """Is this entry file-like?"""740 try:741 return self.info(path)["type"] == "file"742 except: # noqa: E722743 return False744 745 def read_text(self, path, encoding=None, errors=None, newline=None, **kwargs):746 """Get the contents of the file as a string.747 748 Parameters749 ----------750 path: str751 URL of file on this filesystems752 encoding, errors, newline: same as `open`.753 """754 with self.open(755 path,756 mode="r",757 encoding=encoding,758 errors=errors,759 newline=newline,760 **kwargs,761 ) as f:762 return f.read()763 764 def write_text(765 self, path, value, encoding=None, errors=None, newline=None, **kwargs766 ):767 """Write the text to the given file.768 769 An existing file will be overwritten.770 771 Parameters772 ----------773 path: str774 URL of file on this filesystems775 value: str776 Text to write.777 encoding, errors, newline: same as `open`.778 """779 with self.open(780 path,781 mode="w",782 encoding=encoding,783 errors=errors,784 newline=newline,785 **kwargs,786 ) as f:787 return f.write(value)788 789 def cat_file(self, path, start=None, end=None, **kwargs):790 """Get the content of a file791 792 Parameters793 ----------794 path: URL of file on this filesystems795 start, end: int796 Bytes limits of the read. If negative, backwards from end,797 like usual python slices. Either can be None for start or798 end of file, respectively799 kwargs: passed to ``open()``.800 """801 # explicitly set buffering off?802 with self.open(path, "rb", **kwargs) as f:803 if start is not None:804 if start >= 0:805 f.seek(start)806 else:807 f.seek(max(0, f.size + start))808 if end is not None:809 if end < 0:810 end = f.size + end811 return f.read(end - f.tell())812 return f.read()813 814 def pipe_file(self, path, value, mode="overwrite", **kwargs):815 """Set the bytes of given file"""816 if mode == "create" and self.exists(path):817 # non-atomic but simple way; or could use "xb" in open(), which is likely818 # not as well supported819 raise FileExistsError820 with self.open(path, "wb", **kwargs) as f:821 f.write(value)822 823 def pipe(self, path, value=None, **kwargs):824 """Put value into path825 826 (counterpart to ``cat``)827 828 Parameters829 ----------830 path: string or dict(str, bytes)831 If a string, a single remote location to put ``value`` bytes; if a dict,832 a mapping of {path: bytesvalue}.833 value: bytes, optional834 If using a single path, these are the bytes to put there. Ignored if835 ``path`` is a dict836 """837 if isinstance(path, str):838 self.pipe_file(self._strip_protocol(path), value, **kwargs)839 elif isinstance(path, dict):840 for k, v in path.items():841 self.pipe_file(self._strip_protocol(k), v, **kwargs)842 else:843 raise ValueError("path must be str or dict")844 845 def cat_ranges(846 self, paths, starts, ends, max_gap=None, on_error="return", **kwargs847 ):848 """Get the contents of byte ranges from one or more files849 850 Parameters851 ----------852 paths: list853 A list of of filepaths on this filesystems854 starts, ends: int or list855 Bytes limits of the read. If using a single int, the same value will be856 used to read all the specified files.857 """858 if max_gap is not None:859 raise NotImplementedError860 if not isinstance(paths, list):861 raise TypeError862 if not isinstance(starts, list):863 starts = [starts] * len(paths)864 if not isinstance(ends, list):865 ends = [ends] * len(paths)866 if len(starts) != len(paths) or len(ends) != len(paths):867 raise ValueError868 out = []869 for p, s, e in zip(paths, starts, ends):870 try:871 out.append(self.cat_file(p, s, e))872 except Exception as e:873 if on_error == "return":874 out.append(e)875 else:876 raise877 return out878 879 def cat(self, path, recursive=False, on_error="raise", **kwargs):880 """Fetch (potentially multiple) paths' contents881 882 Parameters883 ----------884 recursive: bool885 If True, assume the path(s) are directories, and get all the886 contained files887 on_error : "raise", "omit", "return"888 If raise, an underlying exception will be raised (converted to KeyError889 if the type is in self.missing_exceptions); if omit, keys with exception890 will simply not be included in the output; if "return", all keys are891 included in the output, but the value will be bytes or an exception892 instance.893 kwargs: passed to cat_file894 895 Returns896 -------897 dict of {path: contents} if there are multiple paths898 or the path has been otherwise expanded899 """900 paths = self.expand_path(path, recursive=recursive, **kwargs)901 if (902 len(paths) > 1903 or isinstance(path, list)904 or paths[0] != self._strip_protocol(path)905 ):906 out = {}907 for path in paths:908 try:909 out[path] = self.cat_file(path, **kwargs)910 except Exception as e:911 if on_error == "raise":912 raise913 if on_error == "return":914 out[path] = e915 return out916 else:917 return self.cat_file(paths[0], **kwargs)918 919 def get_file(self, rpath, lpath, callback=DEFAULT_CALLBACK, outfile=None, **kwargs):920 """Copy single remote file to local"""921 from .implementations.local import LocalFileSystem922 923 if isfilelike(lpath):924 outfile = lpath925 elif self.isdir(rpath):926 os.makedirs(lpath, exist_ok=True)927 return None928 929 fs = LocalFileSystem(auto_mkdir=True)930 fs.makedirs(fs._parent(lpath), exist_ok=True)931 932 with self.open(rpath, "rb", **kwargs) as f1:933 if outfile is None:934 outfile = open(lpath, "wb")935 936 try:937 callback.set_size(getattr(f1, "size", None))938 data = True939 while data:940 data = f1.read(self.blocksize)941 segment_len = outfile.write(data)942 if segment_len is None:943 segment_len = len(data)944 callback.relative_update(segment_len)945 finally:946 if not isfilelike(lpath):947 outfile.close()948 949 def get(950 self,951 rpath,952 lpath,953 recursive=False,954 callback=DEFAULT_CALLBACK,955 maxdepth=None,956 **kwargs,957 ):958 """Copy file(s) to local.959 960 Copies a specific file or tree of files (if recursive=True). If lpath961 ends with a "/", it will be assumed to be a directory, and target files962 will go within. Can submit a list of paths, which may be glob-patterns963 and will be expanded.964 965 Calls get_file for each source.966 """967 if isinstance(lpath, list) and isinstance(rpath, list):968 # No need to expand paths when both source and destination969 # are provided as lists970 rpaths = rpath971 lpaths = lpath972 else:973 from .implementations.local import (974 LocalFileSystem,975 make_path_posix,976 trailing_sep,977 )978 979 source_is_str = isinstance(rpath, str)980 rpaths = self.expand_path(981 rpath, recursive=recursive, maxdepth=maxdepth, **kwargs982 )983 if source_is_str and (not recursive or maxdepth is not None):984 # Non-recursive glob does not copy directories985 rpaths = [p for p in rpaths if not (trailing_sep(p) or self.isdir(p))]986 if not rpaths:987 return988 989 if isinstance(lpath, str):990 lpath = make_path_posix(lpath)991 992 source_is_file = len(rpaths) == 1993 dest_is_dir = isinstance(lpath, str) and (994 trailing_sep(lpath) or LocalFileSystem().isdir(lpath)995 )996 997 exists = source_is_str and (998 (has_magic(rpath) and source_is_file)999 or (not has_magic(rpath) and dest_is_dir and not trailing_sep(rpath))1000 )1001 lpaths = other_paths(1002 rpaths,1003 lpath,1004 exists=exists,1005 flatten=not source_is_str,1006 )1007 1008 callback.set_size(len(lpaths))1009 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):1010 with callback.branched(rpath, lpath) as child:1011 self.get_file(rpath, lpath, callback=child, **kwargs)1012 1013 def put_file(1014 self, lpath, rpath, callback=DEFAULT_CALLBACK, mode="overwrite", **kwargs1015 ):1016 """Copy single file to remote"""1017 if mode == "create" and self.exists(rpath):1018 raise FileExistsError1019 if os.path.isdir(lpath):1020 self.makedirs(rpath, exist_ok=True)1021 return None1022 1023 with open(lpath, "rb") as f1:1024 size = f1.seek(0, 2)1025 callback.set_size(size)1026 f1.seek(0)1027 1028 self.mkdirs(self._parent(os.fspath(rpath)), exist_ok=True)1029 with self.open(rpath, "wb", **kwargs) as f2:1030 while f1.tell() < size:1031 data = f1.read(self.blocksize)1032 segment_len = f2.write(data)1033 if segment_len is None:1034 segment_len = len(data)1035 callback.relative_update(segment_len)1036 1037 def put(1038 self,1039 lpath,1040 rpath,1041 recursive=False,1042 callback=DEFAULT_CALLBACK,1043 maxdepth=None,1044 **kwargs,1045 ):1046 """Copy file(s) from local.1047 1048 Copies a specific file or tree of files (if recursive=True). If rpath1049 ends with a "/", it will be assumed to be a directory, and target files1050 will go within.1051 1052 Calls put_file for each source.1053 """1054 if isinstance(lpath, list) and isinstance(rpath, list):1055 # No need to expand paths when both source and destination1056 # are provided as lists1057 rpaths = rpath1058 lpaths = lpath1059 else:1060 from .implementations.local import (1061 LocalFileSystem,1062 make_path_posix,1063 trailing_sep,1064 )1065 1066 source_is_str = isinstance(lpath, str)1067 if source_is_str:1068 lpath = make_path_posix(lpath)1069 fs = LocalFileSystem()1070 lpaths = fs.expand_path(1071 lpath, recursive=recursive, maxdepth=maxdepth, **kwargs1072 )1073 if source_is_str and (not recursive or maxdepth is not None):1074 # Non-recursive glob does not copy directories1075 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]1076 if not lpaths:1077 return1078 1079 source_is_file = len(lpaths) == 11080 dest_is_dir = isinstance(rpath, str) and (1081 trailing_sep(rpath) or self.isdir(rpath)1082 )1083 1084 rpath = (1085 self._strip_protocol(rpath)1086 if isinstance(rpath, str)1087 else [self._strip_protocol(p) for p in rpath]1088 )1089 exists = source_is_str and (1090 (has_magic(lpath) and source_is_file)1091 or (not has_magic(lpath) and dest_is_dir and not trailing_sep(lpath))1092 )1093 rpaths = other_paths(1094 lpaths,1095 rpath,1096 exists=exists,1097 flatten=not source_is_str,1098 )1099 1100 callback.set_size(len(rpaths))1101 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):1102 with callback.branched(lpath, rpath) as child:1103 self.put_file(lpath, rpath, callback=child, **kwargs)1104 1105 def head(self, path, size=1024):1106 """Get the first ``size`` bytes from file"""1107 with self.open(path, "rb") as f:1108 return f.read(size)1109 1110 def tail(self, path, size=1024):1111 """Get the last ``size`` bytes from file"""1112 with self.open(path, "rb") as f:1113 f.seek(max(-size, -f.size), 2)1114 return f.read()1115 1116 def cp_file(self, path1, path2, **kwargs):1117 raise NotImplementedError1118 1119 def copy(1120 self, path1, path2, recursive=False, maxdepth=None, on_error=None, **kwargs1121 ):1122 """Copy within two locations in the filesystem1123 1124 on_error : "raise", "ignore"1125 If raise, any not-found exceptions will be raised; if ignore any1126 not-found exceptions will cause the path to be skipped; defaults to1127 raise unless recursive is true, where the default is ignore1128 """1129 if on_error is None and recursive:1130 on_error = "ignore"1131 elif on_error is None:1132 on_error = "raise"1133 1134 if isinstance(path1, list) and isinstance(path2, list):1135 # No need to expand paths when both source and destination1136 # are provided as lists1137 paths1 = path11138 paths2 = path21139 else:1140 from .implementations.local import trailing_sep1141 1142 source_is_str = isinstance(path1, str)1143 paths1 = self.expand_path(1144 path1, recursive=recursive, maxdepth=maxdepth, **kwargs1145 )1146 if source_is_str and (not recursive or maxdepth is not None):1147 # Non-recursive glob does not copy directories1148 paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))]1149 if not paths1:1150 return1151 1152 source_is_file = len(paths1) == 11153 dest_is_dir = isinstance(path2, str) and (1154 trailing_sep(path2) or self.isdir(path2)1155 )1156 1157 exists = source_is_str and (1158 (has_magic(path1) and source_is_file)1159 or (not has_magic(path1) and dest_is_dir and not trailing_sep(path1))1160 )1161 paths2 = other_paths(1162 paths1,1163 path2,1164 exists=exists,1165 flatten=not source_is_str,1166 )1167 1168 for p1, p2 in zip(paths1, paths2):1169 try:1170 self.cp_file(p1, p2, **kwargs)1171 except FileNotFoundError:1172 if on_error == "raise":1173 raise1174 1175 def expand_path(self, path, recursive=False, maxdepth=None, **kwargs):1176 """Turn one or more globs or directories into a list of all matching paths1177 to files or directories.1178 1179 kwargs are passed to ``glob`` or ``find``, which may in turn call ``ls``1180 """1181 1182 if maxdepth is not None and maxdepth < 1:1183 raise ValueError("maxdepth must be at least 1")1184 1185 if isinstance(path, (str, os.PathLike)):1186 out = self.expand_path([path], recursive, maxdepth, **kwargs)1187 else:1188 out = set()1189 path = [self._strip_protocol(p) for p in path]1190 for p in path:1191 if has_magic(p):1192 bit = set(self.glob(p, maxdepth=maxdepth, **kwargs))1193 out |= bit1194 if recursive:1195 # glob call above expanded one depth so if maxdepth is defined1196 # then decrement it in expand_path call below. If it is zero1197 # after decrementing then avoid expand_path call.1198 if maxdepth is not None and maxdepth <= 1:1199 continue1200 out |= set(