Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fsspec/spec.py: 25%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import io
4import json
5import logging
6import os
7import threading
8import warnings
9import weakref
10from errno import ESPIPE
11from glob import has_magic
12from hashlib import sha256
13from typing import Any, ClassVar
15from .callbacks import DEFAULT_CALLBACK
16from .config import apply_config, conf
17from .dircache import DirCache
18from .transaction import Transaction
19from .utils import (
20 _unstrip_protocol,
21 glob_translate,
22 isfilelike,
23 other_paths,
24 read_block,
25 stringify_path,
26 tokenize,
27)
29logger = logging.getLogger("fsspec")
32def make_instance(cls, args, kwargs):
33 return cls(*args, **kwargs)
36FORK_AVAILABLE = hasattr(os, "register_at_fork")
39if FORK_AVAILABLE:
40 _registered_classes = weakref.WeakSet()
42 def _reset_instances_lock():
43 for cls in _registered_classes:
44 cls._instantiation_lock = threading.RLock()
45 cls._cache.clear()
46 cls._pid = os.getpid()
48 os.register_at_fork(after_in_child=_reset_instances_lock)
51class _Cached(type):
52 """
53 Metaclass for caching file system instances.
55 Notes
56 -----
57 Instances are cached according to
59 * The values of the class attributes listed in `_extra_tokenize_attributes`
60 * The arguments passed to ``__init__``.
62 This creates an additional reference to the filesystem, which prevents the
63 filesystem from being garbage collected when all *user* references go away.
64 A call to the :meth:`AbstractFileSystem.clear_instance_cache` must *also*
65 be made for a filesystem instance to be garbage collected.
66 """
68 def __init__(cls, *args, **kwargs):
69 super().__init__(*args, **kwargs)
71 # Note: we intentionally create a reference here, to avoid garbage
72 # collecting instances when all other references are gone. To really
73 # delete a FileSystem, the cache must be cleared.
74 if conf.get("weakref_instance_cache"): # pragma: no cover
75 # debug option for analysing fork/spawn conditions
76 cls._cache = weakref.WeakValueDictionary()
77 else:
78 cls._cache = {}
79 cls._pid = os.getpid()
80 cls._instantiation_lock = threading.RLock()
82 if FORK_AVAILABLE:
83 _registered_classes.add(cls)
85 def _check_instance_cache(cls, token):
86 inst = cls._cache.get(token)
87 if inst is not None:
88 cls._latest = token
89 return inst
91 def __call__(cls, *args, **kwargs):
92 kwargs = apply_config(cls, kwargs)
93 extra_tokens = tuple(
94 getattr(cls, attr, None) for attr in cls._extra_tokenize_attributes
95 )
96 strip_tokenize_options = {
97 k: kwargs.pop(k) for k in cls._strip_tokenize_options if k in kwargs
98 }
99 pid = os.getpid()
101 if getattr(cls, "async_impl", False) and not kwargs.get("asynchronous", False):
102 token = tokenize(cls, pid, *args, *extra_tokens, **kwargs)
103 else:
104 token = tokenize(
105 cls, pid, threading.get_ident(), *args, *extra_tokens, **kwargs
106 )
107 skip = kwargs.pop("skip_instance_cache", False)
109 if pid != cls._pid:
110 with cls._instantiation_lock:
111 if pid != cls._pid:
112 cls._cache.clear()
113 cls._pid = pid
115 if not skip and cls.cachable:
116 inst = cls._check_instance_cache(token)
117 if inst is not None:
118 return inst
120 with cls._instantiation_lock:
121 # protect against the race condition that a new instance was created
122 # and inserted into the cache since the initial check just above
123 inst = cls._check_instance_cache(token)
124 if inst is not None:
125 return inst
127 obj = super().__call__(*args, **kwargs, **strip_tokenize_options)
128 # Setting _fs_token here causes some static linters to complain.
129 obj._fs_token_ = token
130 obj.storage_args = args
131 obj.storage_options = kwargs
132 if obj.async_impl and obj.mirror_sync_methods:
133 from .asyn import mirror_sync_methods
135 mirror_sync_methods(obj)
137 if cls.cachable and not skip:
138 with cls._instantiation_lock:
139 # another thread may have created the instance while we were calling
140 # super().__call__(), so we check again.
141 inst = cls._check_instance_cache(token)
142 if inst is not None:
143 return inst
145 cls._latest = token
146 cls._cache[token] = obj
147 return obj
150class AbstractFileSystem(metaclass=_Cached):
151 """
152 An abstract super-class for pythonic file-systems
154 Implementations are expected to be compatible with or, better, subclass
155 from here.
156 """
158 cachable = True # this class can be cached, instances reused
159 _cached = False
160 blocksize = 2**22
161 sep = "/"
162 protocol: ClassVar[str | tuple[str, ...]] = "abstract"
163 _latest = None
164 async_impl = False
165 mirror_sync_methods = False
166 root_marker = "" # For some FSs, may require leading '/' or other character
167 transaction_type = Transaction
169 #: Extra *class attributes* that should be considered when hashing.
170 _extra_tokenize_attributes = ()
171 #: *storage options* that should not be considered when hashing.
172 _strip_tokenize_options = ()
174 # Set by _Cached metaclass
175 storage_args: tuple[Any, ...]
176 storage_options: dict[str, Any]
178 def __init__(self, *args, **storage_options):
179 """Create and configure file-system instance
181 Instances may be cachable, so if similar enough arguments are seen
182 a new instance is not required. The token attribute exists to allow
183 implementations to cache instances if they wish.
185 A reasonable default should be provided if there are no arguments.
187 Subclasses should call this method.
189 Parameters
190 ----------
191 use_listings_cache, listings_expiry_time, max_paths:
192 passed to ``DirCache``, if the implementation supports
193 directory listing caching. Pass use_listings_cache=False
194 to disable such caching.
195 skip_instance_cache: bool
196 If this is a cachable implementation, pass True here to force
197 creating a new instance even if a matching instance exists, and prevent
198 storing this instance.
199 asynchronous: bool
200 loop: asyncio-compatible IOLoop or None
201 """
202 if self._cached:
203 # reusing instance, don't change
204 return
205 self._cached = True
206 self._intrans = False
207 self._transaction = None
208 self._invalidated_caches_in_transaction = []
209 self.dircache = DirCache(**storage_options)
211 if storage_options.pop("add_docs", None):
212 warnings.warn("add_docs is no longer supported.", FutureWarning)
214 if storage_options.pop("add_aliases", None):
215 warnings.warn("add_aliases has been removed.", FutureWarning)
216 # This is set in _Cached
217 self._fs_token_ = None
219 @property
220 def fsid(self):
221 """Persistent filesystem id that can be used to compare filesystems
222 across sessions.
223 """
224 raise NotImplementedError
226 @property
227 def _fs_token(self):
228 return self._fs_token_
230 def __dask_tokenize__(self):
231 return self._fs_token
233 def __hash__(self):
234 return int(self._fs_token, 16)
236 def __eq__(self, other):
237 return isinstance(other, type(self)) and self._fs_token == other._fs_token
239 def __reduce__(self):
240 return make_instance, (type(self), self.storage_args, self.storage_options)
242 @classmethod
243 def _strip_protocol(cls, path):
244 """Turn path from fully-qualified to file-system-specific
246 May require FS-specific handling, e.g., for relative paths or links.
247 """
248 if isinstance(path, list):
249 return [cls._strip_protocol(p) for p in path]
250 path = stringify_path(path)
251 protos = (cls.protocol,) if isinstance(cls.protocol, str) else cls.protocol
252 for protocol in protos:
253 if path.startswith(protocol + "://"):
254 path = path[len(protocol) + 3 :]
255 elif path.startswith(protocol + "::"):
256 path = path[len(protocol) + 2 :]
257 path = path.rstrip("/")
258 # use of root_marker to make minimum required path, e.g., "/"
259 return path or cls.root_marker
261 def unstrip_protocol(self, name: str) -> str:
262 """Format FS-specific path to generic, including protocol"""
263 protos = (self.protocol,) if isinstance(self.protocol, str) else self.protocol
264 for protocol in protos:
265 if name.startswith(f"{protocol}://"):
266 return name
267 return f"{protos[0]}://{name}"
269 @staticmethod
270 def _get_kwargs_from_urls(path):
271 """If kwargs can be encoded in the paths, extract them here
273 This should happen before instantiation of the class; incoming paths
274 then should be amended to strip the options in methods.
276 Examples may look like an sftp path "sftp://user@host:/my/path", where
277 the user and host should become kwargs and later get stripped.
278 """
279 # by default, nothing happens
280 return {}
282 @classmethod
283 def current(cls):
284 """Return the most recently instantiated FileSystem
286 If no instance has been created, then create one with defaults
287 """
288 inst = cls._cache.get(cls._latest)
289 if inst is not None:
290 return inst
291 return cls()
293 @property
294 def transaction(self):
295 """A context within which files are committed together upon exit
297 Requires the file class to implement `.commit()` and `.discard()`
298 for the normal and exception cases.
299 """
300 if self._transaction is None:
301 self._transaction = self.transaction_type(self)
302 return self._transaction
304 def start_transaction(self):
305 """Begin write transaction for deferring files, non-context version"""
306 self._intrans = True
307 self._transaction = self.transaction_type(self)
308 return self.transaction
310 def end_transaction(self):
311 """Finish write transaction, non-context version"""
312 self.transaction.complete()
313 self._transaction = None
314 # The invalid cache must be cleared after the transaction is completed.
315 for path in self._invalidated_caches_in_transaction:
316 self.invalidate_cache(path)
317 self._invalidated_caches_in_transaction.clear()
319 def invalidate_cache(self, path=None):
320 """
321 Discard any cached directory information
323 Parameters
324 ----------
325 path: string or None
326 If None, clear all listings cached else listings at or under given
327 path.
328 """
329 # Not necessary to implement invalidation mechanism, may have no cache.
330 # But if have, you should call this method of parent class from your
331 # subclass to ensure expiring caches after transacations correctly.
332 # See the implementation of FTPFileSystem in ftp.py
333 if self._intrans:
334 self._invalidated_caches_in_transaction.append(path)
336 def mkdir(self, path, create_parents=True, **kwargs):
337 """
338 Create directory entry at path
340 For systems that don't have true directories, may create an for
341 this instance only and not touch the real filesystem
343 Parameters
344 ----------
345 path: str
346 location
347 create_parents: bool
348 if True, this is equivalent to ``makedirs``
349 kwargs:
350 may be permissions, etc.
351 """
352 pass # not necessary to implement, may not have directories
354 def makedirs(self, path, exist_ok=False):
355 """Recursively make directories
357 Creates directory at path and any intervening required directories.
358 Raises exception if, for instance, the path already exists but is a
359 file.
361 Parameters
362 ----------
363 path: str
364 leaf directory name
365 exist_ok: bool (False)
366 If False, will error if the target already exists
367 """
368 pass # not necessary to implement, may not have directories
370 def rmdir(self, path):
371 """Remove a directory, if empty"""
372 pass # not necessary to implement, may not have directories
374 def ls(self, path, detail=True, **kwargs):
375 """List objects at path.
377 This should include subdirectories and files at that location. The
378 difference between a file and a directory must be clear when details
379 are requested.
381 The specific keys, or perhaps a FileInfo class, or similar, is TBD,
382 but must be consistent across implementations.
383 Must include:
385 - full path to the entry (without protocol)
386 - size of the entry, in bytes. If the value cannot be determined, will
387 be ``None``.
388 - type of entry, "file", "directory" or other
390 Additional information
391 may be present, appropriate to the file-system, e.g., generation,
392 checksum, etc.
394 May use refresh=True|False to allow use of self._ls_from_cache to
395 check for a saved listing and avoid calling the backend. This would be
396 common where listing may be expensive.
398 Parameters
399 ----------
400 path: str
401 detail: bool
402 if True, gives a list of dictionaries, where each is the same as
403 the result of ``info(path)``. If False, gives a list of paths
404 (str).
405 kwargs: may have additional backend-specific options, such as version
406 information
408 Returns
409 -------
410 List of strings if detail is False, or list of directory information
411 dicts if detail is True.
412 """
413 raise NotImplementedError
415 def _ls_from_cache(self, path):
416 """Check cache for listing
418 Returns listing, if found (may be empty list for a directly that exists
419 but contains nothing), None if not in cache.
420 """
421 parent = self._parent(path)
422 try:
423 return self.dircache[path.rstrip("/")]
424 except KeyError:
425 pass
426 try:
427 files = [
428 f
429 for f in self.dircache[parent]
430 if f["name"] == path
431 or (f["name"] == path.rstrip("/") and f["type"] == "directory")
432 ]
433 if len(files) == 0:
434 # parent dir was listed but did not contain this file
435 raise FileNotFoundError(path)
436 return files
437 except KeyError:
438 pass
440 def walk(self, path, maxdepth=None, topdown=True, on_error="omit", **kwargs):
441 """Return all files under the given path.
443 List all files, recursing into subdirectories; output is iterator-style,
444 like ``os.walk()``. For a simple list of files, ``find()`` is available.
446 When topdown is True, the caller can modify the dirnames list in-place (perhaps
447 using del or slice assignment), and walk() will
448 only recurse into the subdirectories whose names remain in dirnames;
449 this can be used to prune the search, impose a specific order of visiting,
450 or even to inform walk() about directories the caller creates or renames before
451 it resumes walk() again.
452 Modifying dirnames when topdown is False has no effect. (see os.walk)
454 Note that the "files" outputted will include anything that is not
455 a directory, such as links.
457 Parameters
458 ----------
459 path: str
460 Root to recurse into
461 maxdepth: int
462 Maximum recursion depth. None means limitless, but not recommended
463 on link-based file-systems.
464 topdown: bool (True)
465 Whether to walk the directory tree from the top downwards or from
466 the bottom upwards.
467 on_error: "omit", "raise", a callable
468 if omit (default), path with exception will simply be empty;
469 If raise, an underlying exception will be raised;
470 if callable, it will be called with a single OSError instance as argument
471 kwargs: passed to ``ls``
472 """
473 if maxdepth is not None and maxdepth < 1:
474 raise ValueError("maxdepth must be at least 1")
476 path = self._strip_protocol(path)
477 full_dirs = {}
478 dirs = {}
479 files = {}
481 detail = kwargs.pop("detail", False)
482 try:
483 listing = self.ls(path, detail=True, **kwargs)
484 except (FileNotFoundError, OSError) as e:
485 if on_error == "raise":
486 raise
487 if callable(on_error):
488 on_error(e)
489 return
491 for info in listing:
492 # each info name must be at least [path]/part , but here
493 # we check also for names like [path]/part/
494 pathname = info["name"].rstrip("/")
495 name = pathname.rsplit("/", 1)[-1]
496 if info["type"] == "directory" and pathname != path:
497 # do not include "self" path
498 full_dirs[name] = pathname
499 dirs[name] = info
500 elif pathname == path:
501 # file-like with same name as give path
502 files[""] = info
503 else:
504 files[name] = info
506 if not detail:
507 dirs = list(dirs)
508 files = list(files)
510 if topdown:
511 # Yield before recursion if walking top down
512 yield path, dirs, files
514 if maxdepth is not None:
515 maxdepth -= 1
516 if maxdepth < 1:
517 if not topdown:
518 yield path, dirs, files
519 return
521 for d in dirs:
522 yield from self.walk(
523 full_dirs[d],
524 maxdepth=maxdepth,
525 detail=detail,
526 topdown=topdown,
527 **kwargs,
528 )
530 if not topdown:
531 # Yield after recursion if walking bottom up
532 yield path, dirs, files
534 def find(self, path, maxdepth=None, withdirs=False, detail=False, **kwargs):
535 """List all files below path.
537 Like posix ``find`` command without conditions
539 Parameters
540 ----------
541 path : str
542 maxdepth: int or None
543 If not None, the maximum number of levels to descend
544 withdirs: bool
545 Whether to include directory paths in the output. This is True
546 when used by glob, but users usually only want files.
547 kwargs are passed to ``ls``.
548 """
549 # TODO: allow equivalent of -name parameter
550 path = self._strip_protocol(path)
551 out = {}
553 # Add the root directory if withdirs is requested
554 # This is needed for posix glob compliance
555 if withdirs and path != "" and self.isdir(path):
556 out[path] = self.info(path)
558 for _, dirs, files in self.walk(path, maxdepth, detail=True, **kwargs):
559 if withdirs:
560 files.update(dirs)
561 out.update({info["name"]: info for name, info in files.items()})
562 if not out and self.isfile(path):
563 # walk works on directories, but find should also return [path]
564 # when path happens to be a file
565 out[path] = {}
566 names = sorted(out)
567 if not detail:
568 return names
569 else:
570 return {name: out[name] for name in names}
572 def du(self, path, total=True, maxdepth=None, withdirs=False, **kwargs):
573 """Space used by files and optionally directories within a path
575 Directory size does not include the size of its contents.
577 Parameters
578 ----------
579 path: str
580 total: bool
581 Whether to sum all the file sizes
582 maxdepth: int or None
583 Maximum number of directory levels to descend, None for unlimited.
584 withdirs: bool
585 Whether to include directory paths in the output.
586 kwargs: passed to ``find``
588 Returns
589 -------
590 Dict of {path: size} if total=False, or int otherwise, where numbers
591 refer to bytes used.
592 """
593 sizes = {}
594 if withdirs and self.isdir(path):
595 # Include top-level directory in output
596 info = self.info(path)
597 sizes[info["name"]] = info["size"]
598 for f in self.find(path, maxdepth=maxdepth, withdirs=withdirs, **kwargs):
599 info = self.info(f)
600 sizes[info["name"]] = info["size"]
601 if total:
602 return sum(sizes.values())
603 else:
604 return sizes
606 def glob(self, path, maxdepth=None, **kwargs):
607 """Find files by glob-matching.
609 Pattern matching capabilities for finding files that match the given pattern.
611 Parameters
612 ----------
613 path: str
614 The glob pattern to match against
615 maxdepth: int or None
616 Maximum depth for ``'**'`` patterns. Applied on the first ``'**'`` found.
617 Must be at least 1 if provided.
618 kwargs:
619 Additional arguments passed to ``find`` (e.g., detail=True)
621 Returns
622 -------
623 List of matched paths, or dict of paths and their info if detail=True
625 Notes
626 -----
627 Supported patterns:
628 - '*': Matches any sequence of characters within a single directory level
629 - ``'**'``: Matches any number of directory levels (must be an entire path component)
630 - '?': Matches exactly one character
631 - '[abc]': Matches any character in the set
632 - '[a-z]': Matches any character in the range
633 - '[!abc]': Matches any character NOT in the set
635 Special behaviors:
636 - If the path ends with '/', only folders are returned
637 - Consecutive '*' characters are compressed into a single '*'
638 - Empty set '[]' or negated empty negated set '[!]' never match anything
639 - Special characters in character classes are escaped properly
641 Limitations:
642 - ``'**'`` must be a complete path component (e.g., ``'a/**/b'``, not ``'a**b'``)
643 - No brace expansion ('{a,b}.txt')
644 - No extended glob patterns ('+(pattern)', '!(pattern)')
645 """
646 if maxdepth is not None and maxdepth < 1:
647 raise ValueError("maxdepth must be at least 1")
649 import re
651 seps = (os.path.sep, os.path.altsep) if os.path.altsep else (os.path.sep,)
652 ends_with_sep = path.endswith(seps) # _strip_protocol strips trailing slash
653 path = self._strip_protocol(path)
654 append_slash_to_dirname = ends_with_sep or path.endswith(
655 tuple(sep + "**" for sep in seps)
656 )
657 idx_star = path.find("*") if path.find("*") >= 0 else len(path)
658 idx_qmark = path.find("?") if path.find("?") >= 0 else len(path)
659 idx_brace = path.find("[") if path.find("[") >= 0 else len(path)
661 min_idx = min(idx_star, idx_qmark, idx_brace)
663 detail = kwargs.pop("detail", False)
664 withdirs = kwargs.pop("withdirs", True)
666 if not has_magic(path):
667 if self.exists(path, **kwargs):
668 if not detail:
669 return [path]
670 else:
671 return {path: self.info(path, **kwargs)}
672 else:
673 if not detail:
674 return [] # glob of non-existent returns empty
675 else:
676 return {}
677 elif "/" in path[:min_idx]:
678 min_idx = path[:min_idx].rindex("/")
679 root = path[: min_idx + 1]
680 depth = path[min_idx + 1 :].count("/") + 1
681 else:
682 root = ""
683 depth = path[min_idx + 1 :].count("/") + 1
685 if "**" in path:
686 if maxdepth is not None:
687 idx_double_stars = path.find("**")
688 depth_double_stars = path[idx_double_stars:].count("/") + 1
689 depth = depth - depth_double_stars + maxdepth
690 else:
691 depth = None
693 allpaths = self.find(
694 root, maxdepth=depth, withdirs=withdirs, detail=True, **kwargs
695 )
697 pattern = glob_translate(path + ("/" if ends_with_sep else ""))
698 pattern = re.compile(pattern)
700 out = {
701 p: info
702 for p, info in sorted(allpaths.items())
703 if pattern.match(
704 p + "/"
705 if append_slash_to_dirname and info["type"] == "directory"
706 else p
707 )
708 }
710 if detail:
711 return out
712 else:
713 return list(out)
715 def exists(self, path, **kwargs):
716 """Is there a file at the given path"""
717 try:
718 self.info(path, **kwargs)
719 return True
720 except: # noqa: E722
721 # any exception allowed bar FileNotFoundError?
722 return False
724 def lexists(self, path, **kwargs):
725 """If there is a file at the given path (including
726 broken links)"""
727 return self.exists(path)
729 def info(self, path, **kwargs):
730 """Give details of entry at path
732 Returns a single dictionary, with exactly the same information as ``ls``
733 would with ``detail=True``.
735 The default implementation calls ls and could be overridden by a
736 shortcut. kwargs are passed on to ```ls()``.
738 Some file systems might not be able to measure the file's size, in
739 which case, the returned dict will include ``'size': None``.
741 Returns
742 -------
743 dict with keys: name (full path in the FS), size (in bytes), type (file,
744 directory, or something else) and other FS-specific keys.
745 """
746 path = self._strip_protocol(path)
747 out = self.ls(self._parent(path), detail=True, **kwargs)
748 out = [o for o in out if o["name"].rstrip("/") == path]
749 if out:
750 return out[0]
751 out = self.ls(path, detail=True, **kwargs)
752 path = path.rstrip("/")
753 out1 = [o for o in out if o["name"].rstrip("/") == path]
754 if len(out1) == 1:
755 if "size" not in out1[0]:
756 out1[0]["size"] = None
757 return out1[0]
758 elif len(out1) > 1 or out:
759 return {"name": path, "size": 0, "type": "directory"}
760 else:
761 raise FileNotFoundError(path)
763 def checksum(self, path):
764 """Unique value for current version of file
766 If the checksum is the same from one moment to another, the contents
767 are guaranteed to be the same. If the checksum changes, the contents
768 *might* have changed.
770 This should normally be overridden; default will probably capture
771 creation/modification timestamp (which would be good) or maybe
772 access timestamp (which would be bad)
773 """
774 return int(tokenize(self.info(path)), 16)
776 def size(self, path):
777 """Size in bytes of file"""
778 return self.info(path).get("size", None)
780 def sizes(self, paths):
781 """Size in bytes of each file in a list of paths"""
782 return [self.size(p) for p in paths]
784 def isdir(self, path):
785 """Is this entry directory-like?"""
786 try:
787 return self.info(path)["type"] == "directory"
788 except OSError:
789 return False
791 def isfile(self, path):
792 """Is this entry file-like?"""
793 try:
794 return self.info(path)["type"] == "file"
795 except: # noqa: E722
796 return False
798 def read_text(self, path, encoding=None, errors=None, newline=None, **kwargs):
799 """Get the contents of the file as a string.
801 Parameters
802 ----------
803 path: str
804 URL of file on this filesystems
805 encoding, errors, newline: same as `open`.
806 """
807 with self.open(
808 path,
809 mode="r",
810 encoding=encoding,
811 errors=errors,
812 newline=newline,
813 **kwargs,
814 ) as f:
815 return f.read()
817 def write_text(
818 self, path, value, encoding=None, errors=None, newline=None, **kwargs
819 ):
820 """Write the text to the given file.
822 An existing file will be overwritten.
824 Parameters
825 ----------
826 path: str
827 URL of file on this filesystems
828 value: str
829 Text to write.
830 encoding, errors, newline: same as `open`.
831 """
832 with self.open(
833 path,
834 mode="w",
835 encoding=encoding,
836 errors=errors,
837 newline=newline,
838 **kwargs,
839 ) as f:
840 return f.write(value)
842 def cat_file(self, path, start=None, end=None, **kwargs):
843 """Get the content of a file
845 Parameters
846 ----------
847 path: URL of file on this filesystems
848 start, end: int
849 Bytes limits of the read. If negative, backwards from end,
850 like usual python slices. Either can be None for start or
851 end of file, respectively
852 kwargs: passed to ``open()``.
853 """
854 # explicitly set buffering off?
855 with self.open(path, "rb", **kwargs) as f:
856 if start is not None:
857 if start >= 0:
858 f.seek(start)
859 else:
860 f.seek(max(0, f.size + start))
861 if end is not None:
862 if end < 0:
863 end = f.size + end
864 return f.read(end - f.tell())
865 return f.read()
867 def pipe_file(self, path, value, mode="overwrite", **kwargs):
868 """Set the bytes of given file"""
869 if mode == "create" and self.exists(path):
870 # non-atomic but simple way; or could use "xb" in open(), which is likely
871 # not as well supported
872 raise FileExistsError
873 with self.open(path, "wb", **kwargs) as f:
874 f.write(value)
876 def pipe(self, path, value=None, **kwargs):
877 """Put value into path
879 (counterpart to ``cat``)
881 Parameters
882 ----------
883 path: string or dict(str, bytes)
884 If a string, a single remote location to put ``value`` bytes; if a dict,
885 a mapping of {path: bytesvalue}.
886 value: bytes, optional
887 If using a single path, these are the bytes to put there. Ignored if
888 ``path`` is a dict
889 """
890 if isinstance(path, str):
891 self.pipe_file(self._strip_protocol(path), value, **kwargs)
892 elif isinstance(path, dict):
893 for k, v in path.items():
894 self.pipe_file(self._strip_protocol(k), v, **kwargs)
895 else:
896 raise ValueError("path must be str or dict")
898 def cat_ranges(
899 self, paths, starts, ends, max_gap=None, on_error="return", **kwargs
900 ):
901 """Get the contents of byte ranges from one or more files
903 Parameters
904 ----------
905 paths: list
906 A list of of filepaths on this filesystems
907 starts, ends: int or list
908 Bytes limits of the read. If using a single int, the same value will be
909 used to read all the specified files.
910 """
911 if max_gap is not None:
912 raise NotImplementedError
913 if not isinstance(paths, list):
914 raise TypeError
915 if not isinstance(starts, list):
916 starts = [starts] * len(paths)
917 if not isinstance(ends, list):
918 ends = [ends] * len(paths)
919 if len(starts) != len(paths) or len(ends) != len(paths):
920 raise ValueError
921 out = []
922 for p, s, e in zip(paths, starts, ends):
923 try:
924 out.append(self.cat_file(p, s, e, **kwargs))
925 except Exception as e:
926 if on_error == "return":
927 out.append(e)
928 else:
929 raise
930 return out
932 def cat(self, path, recursive=False, on_error="raise", **kwargs):
933 """Fetch (potentially multiple) paths' contents
935 Parameters
936 ----------
937 recursive: bool
938 If True, assume the path(s) are directories, and get all the
939 contained files
940 on_error : "raise", "omit", "return"
941 If raise, an underlying exception will be raised (converted to KeyError
942 if the type is in self.missing_exceptions); if omit, keys with exception
943 will simply not be included in the output; if "return", all keys are
944 included in the output, but the value will be bytes or an exception
945 instance.
946 kwargs: passed to cat_file
948 Returns
949 -------
950 dict of {path: contents} if there are multiple paths
951 or the path has been otherwise expanded
952 """
953 paths = self.expand_path(path, recursive=recursive, **kwargs)
954 if (
955 len(paths) > 1
956 or isinstance(path, list)
957 or paths[0] != self._strip_protocol(path)
958 ):
959 out = {}
960 for path in paths:
961 try:
962 out[path] = self.cat_file(path, **kwargs)
963 except Exception as e:
964 if on_error == "raise":
965 raise
966 if on_error == "return":
967 out[path] = e
968 return out
969 else:
970 return self.cat_file(paths[0], **kwargs)
972 def get_file(self, rpath, lpath, callback=DEFAULT_CALLBACK, outfile=None, **kwargs):
973 """Copy single remote file to local"""
974 from .implementations.local import LocalFileSystem
976 if isfilelike(lpath):
977 outfile = lpath
978 elif self.isdir(rpath):
979 os.makedirs(lpath, exist_ok=True)
980 return None
982 fs = LocalFileSystem(auto_mkdir=True)
983 fs.makedirs(fs._parent(lpath), exist_ok=True)
985 with self.open(rpath, "rb", **kwargs) as f1:
986 if outfile is None:
987 outfile = open(lpath, "wb")
989 try:
990 callback.set_size(getattr(f1, "size", None))
991 data = True
992 while data:
993 data = f1.read(self.blocksize)
994 segment_len = outfile.write(data)
995 if segment_len is None:
996 segment_len = len(data)
997 callback.relative_update(segment_len)
998 finally:
999 if not isfilelike(lpath):
1000 outfile.close()
1002 def get(
1003 self,
1004 rpath,
1005 lpath,
1006 recursive=False,
1007 callback=DEFAULT_CALLBACK,
1008 maxdepth=None,
1009 **kwargs,
1010 ):
1011 """Copy file(s) to local.
1013 Copies a specific file or tree of files (if recursive=True). If lpath
1014 ends with a "/", it will be assumed to be a directory, and target files
1015 will go within. Can submit a list of paths, which may be glob-patterns
1016 and will be expanded.
1018 Calls get_file for each source.
1019 """
1020 if isinstance(lpath, list) and isinstance(rpath, list):
1021 # No need to expand paths when both source and destination
1022 # are provided as lists
1023 rpaths = rpath
1024 lpaths = lpath
1025 else:
1026 from .implementations.local import (
1027 LocalFileSystem,
1028 make_path_posix,
1029 trailing_sep,
1030 )
1032 source_is_str = isinstance(rpath, str)
1033 rpaths = self.expand_path(
1034 rpath, recursive=recursive, maxdepth=maxdepth, **kwargs
1035 )
1036 if source_is_str and (not recursive or maxdepth is not None):
1037 # Non-recursive glob does not copy directories
1038 rpaths = [p for p in rpaths if not (trailing_sep(p) or self.isdir(p))]
1039 if not rpaths:
1040 return
1042 if isinstance(lpath, str):
1043 lpath = make_path_posix(lpath)
1045 source_is_file = len(rpaths) == 1
1046 dest_is_dir = isinstance(lpath, str) and (
1047 trailing_sep(lpath) or LocalFileSystem().isdir(lpath)
1048 )
1050 exists = source_is_str and (
1051 (has_magic(rpath) and source_is_file)
1052 or (not has_magic(rpath) and dest_is_dir and not trailing_sep(rpath))
1053 )
1054 lpaths = other_paths(
1055 rpaths,
1056 lpath,
1057 exists=exists,
1058 flatten=not source_is_str,
1059 )
1061 callback.set_size(len(lpaths))
1062 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):
1063 with callback.branched(rpath, lpath) as child:
1064 self.get_file(rpath, lpath, callback=child, **kwargs)
1066 def put_file(
1067 self, lpath, rpath, callback=DEFAULT_CALLBACK, mode="overwrite", **kwargs
1068 ):
1069 """Copy single file to remote"""
1070 if mode == "create" and self.exists(rpath):
1071 raise FileExistsError
1072 if os.path.isdir(lpath):
1073 self.makedirs(rpath, exist_ok=True)
1074 return None
1076 with open(lpath, "rb") as f1:
1077 size = f1.seek(0, 2)
1078 callback.set_size(size)
1079 f1.seek(0)
1081 self.mkdirs(self._parent(os.fspath(rpath)), exist_ok=True)
1082 with self.open(rpath, "wb", **kwargs) as f2:
1083 while f1.tell() < size:
1084 data = f1.read(self.blocksize)
1085 segment_len = f2.write(data)
1086 if segment_len is None:
1087 segment_len = len(data)
1088 callback.relative_update(segment_len)
1090 def put(
1091 self,
1092 lpath,
1093 rpath,
1094 recursive=False,
1095 callback=DEFAULT_CALLBACK,
1096 maxdepth=None,
1097 **kwargs,
1098 ):
1099 """Copy file(s) from local.
1101 Copies a specific file or tree of files (if recursive=True). If rpath
1102 ends with a "/", it will be assumed to be a directory, and target files
1103 will go within.
1105 Calls put_file for each source.
1106 """
1107 if isinstance(lpath, list) and isinstance(rpath, list):
1108 # No need to expand paths when both source and destination
1109 # are provided as lists
1110 rpaths = rpath
1111 lpaths = lpath
1112 else:
1113 from .implementations.local import (
1114 LocalFileSystem,
1115 make_path_posix,
1116 trailing_sep,
1117 )
1119 source_is_str = isinstance(lpath, str)
1120 if source_is_str:
1121 lpath = make_path_posix(lpath)
1122 fs = LocalFileSystem()
1123 lpaths = fs.expand_path(
1124 lpath, recursive=recursive, maxdepth=maxdepth, **kwargs
1125 )
1126 if source_is_str and (not recursive or maxdepth is not None):
1127 # Non-recursive glob does not copy directories
1128 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
1129 if not lpaths:
1130 return
1132 source_is_file = len(lpaths) == 1
1133 dest_is_dir = isinstance(rpath, str) and (
1134 trailing_sep(rpath) or self.isdir(rpath)
1135 )
1137 rpath = (
1138 self._strip_protocol(rpath)
1139 if isinstance(rpath, str)
1140 else [self._strip_protocol(p) for p in rpath]
1141 )
1142 exists = source_is_str and (
1143 (has_magic(lpath) and source_is_file)
1144 or (not has_magic(lpath) and dest_is_dir and not trailing_sep(lpath))
1145 )
1146 rpaths = other_paths(
1147 lpaths,
1148 rpath,
1149 exists=exists,
1150 flatten=not source_is_str,
1151 )
1153 callback.set_size(len(rpaths))
1154 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):
1155 with callback.branched(lpath, rpath) as child:
1156 self.put_file(lpath, rpath, callback=child, **kwargs)
1158 def head(self, path, size=1024):
1159 """Get the first ``size`` bytes from file"""
1160 with self.open(path, "rb") as f:
1161 return f.read(size)
1163 def tail(self, path, size=1024):
1164 """Get the last ``size`` bytes from file"""
1165 with self.open(path, "rb") as f:
1166 f.seek(max(-size, -f.size), 2)
1167 return f.read()
1169 def cp_file(self, path1, path2, **kwargs):
1170 raise NotImplementedError
1172 def copy(
1173 self, path1, path2, recursive=False, maxdepth=None, on_error=None, **kwargs
1174 ):
1175 """Copy within two locations in the filesystem
1177 on_error : "raise", "ignore"
1178 If raise, any not-found exceptions will be raised; if ignore any
1179 not-found exceptions will cause the path to be skipped; defaults to
1180 raise unless recursive is true, where the default is ignore
1181 """
1182 if on_error is None and recursive:
1183 on_error = "ignore"
1184 elif on_error is None:
1185 on_error = "raise"
1187 if isinstance(path1, list) and isinstance(path2, list):
1188 # No need to expand paths when both source and destination
1189 # are provided as lists
1190 paths1 = path1
1191 paths2 = path2
1192 else:
1193 from .implementations.local import trailing_sep
1195 source_is_str = isinstance(path1, str)
1196 paths1 = self.expand_path(
1197 path1, recursive=recursive, maxdepth=maxdepth, **kwargs
1198 )
1199 if source_is_str and (not recursive or maxdepth is not None):
1200 # Non-recursive glob does not copy directories
1201 paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))]
1202 if not paths1:
1203 return
1205 source_is_file = len(paths1) == 1
1206 dest_is_dir = isinstance(path2, str) and (
1207 trailing_sep(path2) or self.isdir(path2)
1208 )
1210 exists = source_is_str and (
1211 (has_magic(path1) and source_is_file)
1212 or (not has_magic(path1) and dest_is_dir and not trailing_sep(path1))
1213 )
1214 paths2 = other_paths(
1215 paths1,
1216 path2,
1217 exists=exists,
1218 flatten=not source_is_str,
1219 )
1221 for p1, p2 in zip(paths1, paths2):
1222 try:
1223 self.cp_file(p1, p2, **kwargs)
1224 except FileNotFoundError:
1225 if on_error == "raise":
1226 raise
1228 def expand_path(
1229 self, path, recursive=False, maxdepth=None, assume_literal=False, **kwargs
1230 ):
1231 """Turn one or more globs or directories into a list of all matching paths
1232 to files or directories.
1234 kwargs are passed to ``glob`` or ``find``, which may in turn call ``ls``
1235 """
1237 if maxdepth is not None and maxdepth < 1:
1238 raise ValueError("maxdepth must be at least 1")
1240 if isinstance(path, (str, os.PathLike)):
1241 out = self.expand_path([path], recursive, maxdepth, **kwargs)
1242 else:
1243 out = set()
1244 path = [self._strip_protocol(p) for p in path]
1245 for p in path:
1246 if not assume_literal and has_magic(p):
1247 bit = set(self.glob(p, maxdepth=maxdepth, **kwargs))
1248 out |= bit
1249 if recursive:
1250 # glob call above expanded one depth so if maxdepth is defined
1251 # then decrement it in expand_path call below. If it is zero
1252 # after decrementing then avoid expand_path call.
1253 if maxdepth is not None and maxdepth <= 1:
1254 continue
1255 out |= set(
1256 self.expand_path(
1257 list(bit),
1258 recursive=recursive,
1259 maxdepth=maxdepth - 1 if maxdepth is not None else None,
1260 assume_literal=True,
1261 **kwargs,
1262 )
1263 )
1264 continue
1265 elif recursive:
1266 rec = set(
1267 self.find(
1268 p, maxdepth=maxdepth, withdirs=True, detail=False, **kwargs
1269 )
1270 )
1271 out |= rec
1272 if p not in out and (recursive is False or self.exists(p)):
1273 # should only check once, for the root
1274 out.add(p)
1275 if not out:
1276 raise FileNotFoundError(path)
1277 return sorted(out)
1279 def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs):
1280 """Move file(s) from one location to another"""
1281 if path1 == path2:
1282 logger.debug("%s mv: The paths are the same, so no files were moved.", self)
1283 else:
1284 # explicitly raise exception to prevent data corruption
1285 self.copy(
1286 path1, path2, recursive=recursive, maxdepth=maxdepth, on_error="raise"
1287 )
1288 self.rm(path1, recursive=recursive)
1290 def rm_file(self, path):
1291 """Delete a file"""
1292 self._rm(path)
1294 def _rm(self, path):
1295 """Delete one file"""
1296 # this is the old name for the method, prefer rm_file
1297 raise NotImplementedError
1299 def rm(self, path, recursive=False, maxdepth=None):
1300 """Delete files.
1302 Parameters
1303 ----------
1304 path: str or list of str
1305 File(s) to delete.
1306 recursive: bool
1307 If file(s) are directories, recursively delete contents and then
1308 also remove the directory
1309 maxdepth: int or None
1310 Depth to pass to walk for finding files to delete, if recursive.
1311 If None, there will be no limit and infinite recursion may be
1312 possible.
1313 """
1314 path = self.expand_path(path, recursive=recursive, maxdepth=maxdepth)
1315 for p in reversed(path):
1316 self.rm_file(p)
1318 @classmethod
1319 def _parent(cls, path):
1320 path = cls._strip_protocol(path)
1321 if "/" in path:
1322 parent = path.rsplit("/", 1)[0].lstrip(cls.root_marker)
1323 return cls.root_marker + parent
1324 else:
1325 return cls.root_marker
1327 def _open(
1328 self,
1329 path,
1330 mode="rb",
1331 block_size=None,
1332 autocommit=True,
1333 cache_options=None,
1334 **kwargs,
1335 ):
1336 """Return raw bytes-mode file-like from the file-system"""
1337 return AbstractBufferedFile(
1338 self,
1339 path,
1340 mode,
1341 block_size,
1342 autocommit,
1343 cache_options=cache_options,
1344 **kwargs,
1345 )
1347 def open(
1348 self,
1349 path,
1350 mode="rb",
1351 block_size=None,
1352 cache_options=None,
1353 compression=None,
1354 **kwargs,
1355 ):
1356 """
1357 Return a file-like object from the filesystem
1359 The resultant instance must function correctly in a context ``with``
1360 block.
1362 Parameters
1363 ----------
1364 path: str
1365 Target file
1366 mode: str like 'rb', 'w'
1367 See builtin ``open()``
1368 Mode "x" (exclusive write) may be implemented by the backend. Even if
1369 it is, whether it is checked up front or on commit, and whether it is
1370 atomic is implementation-dependent.
1371 block_size: int
1372 Some indication of buffering - this is a value in bytes
1373 cache_options : dict, optional
1374 Extra arguments to pass through to the cache.
1375 compression: string or None
1376 If given, open file using compression codec. Can either be a compression
1377 name (a key in ``fsspec.compression.compr``) or "infer" to guess the
1378 compression from the filename suffix.
1379 encoding, errors, newline: passed on to TextIOWrapper for text mode
1380 """
1381 import io
1383 path = self._strip_protocol(path)
1384 if "b" not in mode:
1385 mode = mode.replace("t", "") + "b"
1387 text_kwargs = {
1388 k: kwargs.pop(k)
1389 for k in ["encoding", "errors", "newline"]
1390 if k in kwargs
1391 }
1392 return io.TextIOWrapper(
1393 self.open(
1394 path,
1395 mode,
1396 block_size=block_size,
1397 cache_options=cache_options,
1398 compression=compression,
1399 **kwargs,
1400 ),
1401 **text_kwargs,
1402 )
1403 else:
1404 ac = kwargs.pop("autocommit", not self._intrans)
1405 f = self._open(
1406 path,
1407 mode=mode,
1408 block_size=block_size,
1409 autocommit=ac,
1410 cache_options=cache_options,
1411 **kwargs,
1412 )
1413 if compression is not None:
1414 from fsspec.compression import compr
1415 from fsspec.core import get_compression
1417 compression = get_compression(path, compression)
1418 compress = compr[compression]
1419 f = compress(f, mode=mode[0])
1421 if not ac and "r" not in mode:
1422 self.transaction.files.append(f)
1423 return f
1425 def touch(self, path, truncate=True, **kwargs):
1426 """Create empty file, or update timestamp
1428 Parameters
1429 ----------
1430 path: str
1431 file location
1432 truncate: bool
1433 If True, always set file size to 0; if False, update timestamp and
1434 leave file unchanged, if backend allows this
1435 """
1436 if truncate or not self.exists(path):
1437 with self.open(path, "wb", **kwargs):
1438 pass
1439 else:
1440 raise NotImplementedError # update timestamp, if possible
1442 def ukey(self, path):
1443 """Hash of file properties, to tell if it has changed"""
1444 return sha256(str(self.info(path)).encode()).hexdigest()
1446 def read_block(self, fn, offset, length, delimiter=None):
1447 """Read a block of bytes from
1449 Starting at ``offset`` of the file, read ``length`` bytes. If
1450 ``delimiter`` is set then we ensure that the read starts and stops at
1451 delimiter boundaries that follow the locations ``offset`` and ``offset
1452 + length``. If ``offset`` is zero then we start at zero. The
1453 bytestring returned WILL include the end delimiter string.
1455 If offset+length is beyond the eof, reads to eof.
1457 Parameters
1458 ----------
1459 fn: string
1460 Path to filename
1461 offset: int
1462 Byte offset to start read
1463 length: int
1464 Number of bytes to read. If None, read to end.
1465 delimiter: bytes (optional)
1466 Ensure reading starts and stops at delimiter bytestring
1468 Examples
1469 --------
1470 >>> fs.read_block('data/file.csv', 0, 13) # doctest: +SKIP
1471 b'Alice, 100\\nBo'
1472 >>> fs.read_block('data/file.csv', 0, 13, delimiter=b'\\n') # doctest: +SKIP
1473 b'Alice, 100\\nBob, 200\\n'
1475 Use ``length=None`` to read to the end of the file.
1476 >>> fs.read_block('data/file.csv', 0, None, delimiter=b'\\n') # doctest: +SKIP
1477 b'Alice, 100\\nBob, 200\\nCharlie, 300'
1479 See Also
1480 --------
1481 :func:`fsspec.utils.read_block`
1482 """
1483 with self.open(fn, "rb") as f:
1484 size = f.size
1485 if length is None:
1486 length = size
1487 if size is not None and offset + length > size:
1488 length = size - offset
1489 return read_block(f, offset, length, delimiter)
1491 def to_json(self, *, include_password: bool = True) -> str:
1492 """
1493 JSON representation of this filesystem instance.
1495 Parameters
1496 ----------
1497 include_password: bool, default True
1498 Whether to include the password (if any) in the output.
1500 Returns
1501 -------
1502 JSON string with keys ``cls`` (the python location of this class),
1503 protocol (text name of this class's protocol, first one in case of
1504 multiple), ``args`` (positional args, usually empty), and all other
1505 keyword arguments as their own keys.
1507 Warnings
1508 --------
1509 Serialized filesystems may contain sensitive information which have been
1510 passed to the constructor, such as passwords and tokens. Make sure you
1511 store and send them in a secure environment!
1512 """
1513 from .json import FilesystemJSONEncoder
1515 return json.dumps(
1516 self,
1517 cls=type(
1518 "_FilesystemJSONEncoder",
1519 (FilesystemJSONEncoder,),
1520 {"include_password": include_password},
1521 ),
1522 )
1524 @staticmethod
1525 def from_json(blob: str) -> AbstractFileSystem:
1526 """
1527 Recreate a filesystem instance from JSON representation.
1529 See ``.to_json()`` for the expected structure of the input.
1531 Parameters
1532 ----------
1533 blob: str
1535 Returns
1536 -------
1537 file system instance, not necessarily of this particular class.
1539 Warnings
1540 --------
1541 This can import arbitrary modules (as determined by the ``cls`` key).
1542 Make sure you haven't installed any modules that may execute malicious code
1543 at import time.
1544 """
1545 from .json import FilesystemJSONDecoder
1547 return json.loads(blob, cls=FilesystemJSONDecoder)
1549 def to_dict(self, *, include_password: bool = True) -> dict[str, Any]:
1550 """
1551 JSON-serializable dictionary representation of this filesystem instance.
1553 Parameters
1554 ----------
1555 include_password: bool, default True
1556 Whether to include the password (if any) in the output.
1558 Returns
1559 -------
1560 Dictionary with keys ``cls`` (the python location of this class),
1561 protocol (text name of this class's protocol, first one in case of
1562 multiple), ``args`` (positional args, usually empty), and all other
1563 keyword arguments as their own keys.
1565 Warnings
1566 --------
1567 Serialized filesystems may contain sensitive information which have been
1568 passed to the constructor, such as passwords and tokens. Make sure you
1569 store and send them in a secure environment!
1570 """
1571 from .json import FilesystemJSONEncoder
1573 json_encoder = FilesystemJSONEncoder()
1575 cls = type(self)
1576 proto = self.protocol
1578 storage_options = dict(self.storage_options)
1579 if not include_password:
1580 storage_options.pop("password", None)
1582 return dict(
1583 cls=f"{cls.__module__}:{cls.__name__}",
1584 protocol=proto[0] if isinstance(proto, (tuple, list)) else proto,
1585 args=json_encoder.make_serializable(self.storage_args),
1586 **json_encoder.make_serializable(storage_options),
1587 )
1589 @staticmethod
1590 def from_dict(dct: dict[str, Any]) -> AbstractFileSystem:
1591 """
1592 Recreate a filesystem instance from dictionary representation.
1594 See ``.to_dict()`` for the expected structure of the input.
1596 Parameters
1597 ----------
1598 dct: Dict[str, Any]
1600 Returns
1601 -------
1602 file system instance, not necessarily of this particular class.
1604 Warnings
1605 --------
1606 This can import arbitrary modules (as determined by the ``cls`` key).
1607 Make sure you haven't installed any modules that may execute malicious code
1608 at import time.
1609 """
1610 from .json import FilesystemJSONDecoder
1612 json_decoder = FilesystemJSONDecoder()
1614 dct = dict(dct) # Defensive copy
1616 cls = FilesystemJSONDecoder.try_resolve_fs_cls(dct)
1617 if cls is None:
1618 raise ValueError("Not a serialized AbstractFileSystem")
1620 dct.pop("cls", None)
1621 dct.pop("protocol", None)
1623 return cls(
1624 *json_decoder.unmake_serializable(dct.pop("args", ())),
1625 **json_decoder.unmake_serializable(dct),
1626 )
1628 def _get_pyarrow_filesystem(self):
1629 """
1630 Make a version of the FS instance which will be acceptable to pyarrow
1631 """
1632 # all instances already also derive from pyarrow
1633 return self
1635 def get_mapper(self, root="", check=False, create=False, missing_exceptions=None):
1636 """Create key/value store based on this file-system
1638 Makes a MutableMapping interface to the FS at the given root path.
1639 See ``fsspec.mapping.FSMap`` for further details.
1640 """
1641 from .mapping import FSMap
1643 return FSMap(
1644 root,
1645 self,
1646 check=check,
1647 create=create,
1648 missing_exceptions=missing_exceptions,
1649 )
1651 @classmethod
1652 def clear_instance_cache(cls):
1653 """
1654 Clear the cache of filesystem instances.
1656 Notes
1657 -----
1658 Unless overridden by setting the ``cachable`` class attribute to False,
1659 the filesystem class stores a reference to newly created instances. This
1660 prevents Python's normal rules around garbage collection from working,
1661 since the instances refcount will not drop to zero until
1662 ``clear_instance_cache`` is called.
1663 """
1664 cls._cache.clear()
1666 def created(self, path):
1667 """Return the created timestamp of a file as a datetime.datetime"""
1668 raise NotImplementedError
1670 def modified(self, path):
1671 """Return the modified timestamp of a file as a datetime.datetime"""
1672 raise NotImplementedError
1674 def tree(
1675 self,
1676 path: str = "/",
1677 recursion_limit: int = 2,
1678 max_display: int = 25,
1679 display_size: bool = False,
1680 prefix: str = "",
1681 is_last: bool = True,
1682 first: bool = True,
1683 indent_size: int = 4,
1684 ) -> str:
1685 """
1686 Return a tree-like structure of the filesystem starting from the given path as a string.
1688 Parameters
1689 ----------
1690 path: Root path to start traversal from
1691 recursion_limit: Maximum depth of directory traversal
1692 max_display: Maximum number of items to display per directory
1693 display_size: Whether to display file sizes
1694 prefix: Current line prefix for visual tree structure
1695 is_last: Whether current item is last in its level
1696 first: Whether this is the first call (displays root path)
1697 indent_size: Number of spaces by indent
1699 Returns
1700 -------
1701 str: A string representing the tree structure.
1703 Example
1704 -------
1705 >>> from fsspec import filesystem
1707 >>> fs = filesystem('ftp', host='test.rebex.net', user='demo', password='password')
1708 >>> tree = fs.tree(display_size=True, recursion_limit=3, indent_size=8, max_display=10)
1709 >>> print(tree)
1710 """
1712 def format_bytes(n: int) -> str:
1713 """Format bytes as text."""
1714 for prefix, k in (
1715 ("P", 2**50),
1716 ("T", 2**40),
1717 ("G", 2**30),
1718 ("M", 2**20),
1719 ("k", 2**10),
1720 ):
1721 if n >= 0.9 * k:
1722 return f"{n / k:.2f} {prefix}b"
1723 return f"{n}B"
1725 result = []
1727 if first:
1728 result.append(path)
1730 if recursion_limit:
1731 indent = " " * indent_size
1732 contents = self.ls(path, detail=True)
1733 contents.sort(
1734 key=lambda x: (x.get("type") != "directory", x.get("name", ""))
1735 )
1737 if max_display is not None and len(contents) > max_display:
1738 displayed_contents = contents[:max_display]
1739 remaining_count = len(contents) - max_display
1740 else:
1741 displayed_contents = contents
1742 remaining_count = 0
1744 for i, item in enumerate(displayed_contents):
1745 is_last_item = (i == len(displayed_contents) - 1) and (
1746 remaining_count == 0
1747 )
1749 branch = (
1750 "└" + ("─" * (indent_size - 2))
1751 if is_last_item
1752 else "├" + ("─" * (indent_size - 2))
1753 )
1754 branch += " "
1755 new_prefix = prefix + (
1756 indent if is_last_item else "│" + " " * (indent_size - 1)
1757 )
1759 name = os.path.basename(item.get("name", ""))
1761 if display_size and item.get("type") == "directory":
1762 sub_contents = self.ls(item.get("name", ""), detail=True)
1763 num_files = sum(
1764 1 for sub_item in sub_contents if sub_item.get("type") == "file"
1765 )
1766 num_folders = sum(
1767 1
1768 for sub_item in sub_contents
1769 if sub_item.get("type") == "directory"
1770 )
1772 if num_files == 0 and num_folders == 0:
1773 size = " (empty folder)"
1774 elif num_files == 0:
1775 size = f" ({num_folders} subfolder{'s' if num_folders > 1 else ''})"
1776 elif num_folders == 0:
1777 size = f" ({num_files} file{'s' if num_files > 1 else ''})"
1778 else:
1779 size = f" ({num_files} file{'s' if num_files > 1 else ''}, {num_folders} subfolder{'s' if num_folders > 1 else ''})"
1780 elif display_size and item.get("type") == "file":
1781 size = f" ({format_bytes(item.get('size', 0))})"
1782 else:
1783 size = ""
1785 result.append(f"{prefix}{branch}{name}{size}")
1787 if item.get("type") == "directory" and recursion_limit > 0:
1788 result.append(
1789 self.tree(
1790 path=item.get("name", ""),
1791 recursion_limit=recursion_limit - 1,
1792 max_display=max_display,
1793 display_size=display_size,
1794 prefix=new_prefix,
1795 is_last=is_last_item,
1796 first=False,
1797 indent_size=indent_size,
1798 )
1799 )
1801 if remaining_count > 0:
1802 more_message = f"{remaining_count} more item(s) not displayed."
1803 result.append(
1804 f"{prefix}{'└' + ('─' * (indent_size - 2))} {more_message}"
1805 )
1807 return "\n".join(_ for _ in result if _)
1809 # ------------------------------------------------------------------------
1810 # Aliases
1812 def read_bytes(self, path, start=None, end=None, **kwargs):
1813 """Alias of `AbstractFileSystem.cat_file`."""
1814 return self.cat_file(path, start=start, end=end, **kwargs)
1816 def write_bytes(self, path, value, **kwargs):
1817 """Alias of `AbstractFileSystem.pipe_file`."""
1818 self.pipe_file(path, value, **kwargs)
1820 def makedir(self, path, create_parents=True, **kwargs):
1821 """Alias of `AbstractFileSystem.mkdir`."""
1822 return self.mkdir(path, create_parents=create_parents, **kwargs)
1824 def mkdirs(self, path, exist_ok=False):
1825 """Alias of `AbstractFileSystem.makedirs`."""
1826 return self.makedirs(path, exist_ok=exist_ok)
1828 def listdir(self, path, detail=True, **kwargs):
1829 """Alias of `AbstractFileSystem.ls`."""
1830 return self.ls(path, detail=detail, **kwargs)
1832 def cp(self, path1, path2, **kwargs):
1833 """Alias of `AbstractFileSystem.copy`."""
1834 return self.copy(path1, path2, **kwargs)
1836 def move(self, path1, path2, **kwargs):
1837 """Alias of `AbstractFileSystem.mv`."""
1838 return self.mv(path1, path2, **kwargs)
1840 def stat(self, path, **kwargs):
1841 """Alias of `AbstractFileSystem.info`."""
1842 return self.info(path, **kwargs)
1844 def disk_usage(self, path, total=True, maxdepth=None, **kwargs):
1845 """Alias of `AbstractFileSystem.du`."""
1846 return self.du(path, total=total, maxdepth=maxdepth, **kwargs)
1848 def rename(self, path1, path2, **kwargs):
1849 """Alias of `AbstractFileSystem.mv`."""
1850 return self.mv(path1, path2, **kwargs)
1852 def delete(self, path, recursive=False, maxdepth=None):
1853 """Alias of `AbstractFileSystem.rm`."""
1854 return self.rm(path, recursive=recursive, maxdepth=maxdepth)
1856 def upload(self, lpath, rpath, recursive=False, **kwargs):
1857 """Alias of `AbstractFileSystem.put`."""
1858 return self.put(lpath, rpath, recursive=recursive, **kwargs)
1860 def download(self, rpath, lpath, recursive=False, **kwargs):
1861 """Alias of `AbstractFileSystem.get`."""
1862 return self.get(rpath, lpath, recursive=recursive, **kwargs)
1864 def sign(self, path, expiration=100, **kwargs):
1865 """Create a signed URL representing the given path
1867 Some implementations allow temporary URLs to be generated, as a
1868 way of delegating credentials.
1870 Parameters
1871 ----------
1872 path : str
1873 The path on the filesystem
1874 expiration : int
1875 Number of seconds to enable the URL for (if supported)
1877 Returns
1878 -------
1879 URL : str
1880 The signed URL
1882 Raises
1883 ------
1884 NotImplementedError : if method is not implemented for a filesystem
1885 """
1886 raise NotImplementedError("Sign is not implemented for this filesystem")
1888 def _isfilestore(self):
1889 # Originally inherited from pyarrow DaskFileSystem. Keeping this
1890 # here for backwards compatibility as long as pyarrow uses its
1891 # legacy fsspec-compatible filesystems and thus accepts fsspec
1892 # filesystems as well
1893 return False
1896class AbstractBufferedFile(io.IOBase):
1897 """Convenient class to derive from to provide buffering
1899 In the case that the backend does not provide a pythonic file-like object
1900 already, this class contains much of the logic to build one. The only
1901 methods that need to be overridden are ``_upload_chunk``,
1902 ``_initiate_upload`` and ``_fetch_range``.
1903 """
1905 DEFAULT_BLOCK_SIZE = 5 * 2**20
1906 _details = None
1908 def __init__(
1909 self,
1910 fs,
1911 path,
1912 mode="rb",
1913 block_size="default",
1914 autocommit=True,
1915 cache_type="readahead",
1916 cache_options=None,
1917 size=None,
1918 **kwargs,
1919 ):
1920 """
1921 Template for files with buffered reading and writing
1923 Parameters
1924 ----------
1925 fs: instance of FileSystem
1926 path: str
1927 location in file-system
1928 mode: str
1929 Normal file modes. Currently only 'wb', 'ab' or 'rb'. Some file
1930 systems may be read-only, and some may not support append.
1931 block_size: int
1932 Buffer size for reading or writing, 'default' for class default
1933 autocommit: bool
1934 Whether to write to final destination; may only impact what
1935 happens when file is being closed.
1936 cache_type: {"readahead", "none", "mmap", "bytes"}, default "readahead"
1937 Caching policy in read mode. See the definitions in ``core``.
1938 cache_options : dict
1939 Additional options passed to the constructor for the cache specified
1940 by `cache_type`.
1941 size: int
1942 If given and in read mode, suppressed having to look up the file size
1943 kwargs:
1944 Gets stored as self.kwargs
1945 """
1946 from .core import caches
1948 self.path = path
1949 self.fs = fs
1950 self.mode = mode
1951 self.blocksize = (
1952 self.DEFAULT_BLOCK_SIZE if block_size in ["default", None] else block_size
1953 )
1954 self.loc = 0
1955 self.autocommit = autocommit
1956 self.end = None
1957 self.start = None
1958 self.closed = False
1960 if cache_options is None:
1961 cache_options = {}
1963 if "trim" in kwargs:
1964 warnings.warn(
1965 "Passing 'trim' to control the cache behavior has been deprecated. "
1966 "Specify it within the 'cache_options' argument instead.",
1967 FutureWarning,
1968 )
1969 cache_options["trim"] = kwargs.pop("trim")
1971 self.kwargs = kwargs
1973 if mode not in {"ab", "rb", "wb", "xb"}:
1974 raise NotImplementedError("File mode not supported")
1975 if mode == "rb":
1976 if size is not None:
1977 self.size = size
1978 else:
1979 self.size = self.details["size"]
1980 self.cache = caches[cache_type](
1981 self.blocksize, self._fetch_range, self.size, **cache_options
1982 )
1983 else:
1984 self.buffer = io.BytesIO()
1985 self.offset = None
1986 self.forced = False
1987 self.location = None
1989 @property
1990 def details(self):
1991 if self._details is None:
1992 self._details = self.fs.info(self.path)
1993 return self._details
1995 @details.setter
1996 def details(self, value):
1997 self._details = value
1998 self.size = value["size"]
2000 @property
2001 def full_name(self):
2002 return _unstrip_protocol(self.path, self.fs)
2004 @property
2005 def closed(self):
2006 # get around this attr being read-only in IOBase
2007 # use getattr here, since this can be called during del
2008 return getattr(self, "_closed", True)
2010 @closed.setter
2011 def closed(self, c):
2012 self._closed = c
2014 def __hash__(self):
2015 if "w" in self.mode:
2016 return id(self)
2017 else:
2018 return int(tokenize(self.details), 16)
2020 def __eq__(self, other):
2021 """Files are equal if they have the same checksum, only in read mode"""
2022 if self is other:
2023 return True
2024 return (
2025 isinstance(other, type(self))
2026 and self.mode == "rb"
2027 and other.mode == "rb"
2028 and hash(self) == hash(other)
2029 )
2031 def commit(self):
2032 """Move from temp to final destination"""
2034 def discard(self):
2035 """Throw away temporary file"""
2037 def info(self):
2038 """File information about this path"""
2039 if self.readable():
2040 return self.details
2041 else:
2042 raise ValueError("Info not available while writing")
2044 def tell(self):
2045 """Current file location"""
2046 return self.loc
2048 def seek(self, loc, whence=0):
2049 """Set current file location
2051 Parameters
2052 ----------
2053 loc: int
2054 byte location
2055 whence: {0, 1, 2}
2056 from start of file, current location or end of file, resp.
2057 """
2058 loc = int(loc)
2059 if not self.mode == "rb":
2060 raise OSError(ESPIPE, "Seek only available in read mode")
2061 if whence == 0:
2062 nloc = loc
2063 elif whence == 1:
2064 nloc = self.loc + loc
2065 elif whence == 2:
2066 nloc = self.size + loc
2067 else:
2068 raise ValueError(f"invalid whence ({whence}, should be 0, 1 or 2)")
2069 if nloc < 0:
2070 raise ValueError("Seek before start of file")
2071 self.loc = nloc
2072 return self.loc
2074 def write(self, data):
2075 """
2076 Write data to buffer.
2078 Buffer only sent on flush() or if buffer is greater than
2079 or equal to blocksize.
2081 Parameters
2082 ----------
2083 data: bytes
2084 Set of bytes to be written.
2085 """
2086 if not self.writable():
2087 raise ValueError("File not in write mode")
2088 if self.closed:
2089 raise ValueError("I/O operation on closed file.")
2090 if self.forced:
2091 raise ValueError("This file has been force-flushed, can only close")
2092 out = self.buffer.write(data)
2093 self.loc += out
2094 if self.buffer.tell() >= self.blocksize:
2095 self.flush()
2096 return out
2098 def flush(self, force=False):
2099 """
2100 Write buffered data to backend store.
2102 Writes the current buffer, if it is larger than the block-size, or if
2103 the file is being closed.
2105 Parameters
2106 ----------
2107 force: bool
2108 When closing, write the last block even if it is smaller than
2109 blocks are allowed to be. Disallows further writing to this file.
2110 """
2112 if self.closed:
2113 raise ValueError("Flush on closed file")
2114 if force and self.forced:
2115 raise ValueError("Force flush cannot be called more than once")
2116 if force:
2117 self.forced = True
2119 if self.readable():
2120 # no-op to flush on read-mode
2121 return
2123 if not force and self.buffer.tell() < self.blocksize:
2124 # Defer write on small block
2125 return
2127 if self.offset is None:
2128 # Initialize a multipart upload
2129 self.offset = 0
2130 try:
2131 self._initiate_upload()
2132 except Exception:
2133 self.closed = True
2134 raise
2136 if self._upload_chunk(final=force) is not False:
2137 self.offset += self.buffer.seek(0, 2)
2138 self.buffer = io.BytesIO()
2140 def _upload_chunk(self, final=False):
2141 """Write one part of a multi-block file upload
2143 Parameters
2144 ==========
2145 final: bool
2146 This is the last block, so should complete file, if
2147 self.autocommit is True.
2148 """
2149 # may not yet have been initialized, may need to call _initialize_upload
2151 def _initiate_upload(self):
2152 """Create remote file/upload"""
2153 pass
2155 def _fetch_range(self, start, end):
2156 """Get the specified set of bytes from remote"""
2157 return self.fs.cat_file(self.path, start=start, end=end)
2159 def read(self, length=-1):
2160 """
2161 Return data from cache, or fetch pieces as necessary
2163 Parameters
2164 ----------
2165 length: int (-1)
2166 Number of bytes to read; if <0, all remaining bytes.
2167 """
2168 length = -1 if length is None else int(length)
2169 if self.mode != "rb":
2170 raise ValueError("File not in read mode")
2171 if length < 0:
2172 length = self.size - self.loc
2173 if self.closed:
2174 raise ValueError("I/O operation on closed file.")
2175 if length == 0:
2176 # don't even bother calling fetch
2177 return b""
2178 out = self.cache._fetch(self.loc, self.loc + length)
2180 logger.debug(
2181 "%s read: %i - %i %s",
2182 self,
2183 self.loc,
2184 self.loc + length,
2185 self.cache._log_stats(),
2186 )
2187 self.loc += len(out)
2188 return out
2190 def readinto(self, b):
2191 """mirrors builtin file's readinto method
2193 https://docs.python.org/3/library/io.html#io.RawIOBase.readinto
2194 """
2195 out = memoryview(b).cast("B")
2196 data = self.read(out.nbytes)
2197 out[: len(data)] = data
2198 return len(data)
2200 def readuntil(self, char=b"\n", blocks=None):
2201 """Return data between current position and first occurrence of char
2203 char is included in the output, except if the end of the tile is
2204 encountered first.
2206 Parameters
2207 ----------
2208 char: bytes
2209 Thing to find
2210 blocks: None or int
2211 How much to read in each go. Defaults to file blocksize - which may
2212 mean a new read on every call.
2213 """
2214 out = []
2215 while True:
2216 start = self.tell()
2217 part = self.read(blocks or self.blocksize)
2218 if len(part) == 0:
2219 break
2220 found = part.find(char)
2221 if found > -1:
2222 out.append(part[: found + len(char)])
2223 self.seek(start + found + len(char))
2224 break
2225 out.append(part)
2226 return b"".join(out)
2228 def readline(self):
2229 """Read until and including the first occurrence of newline character
2231 Note that, because of character encoding, this is not necessarily a
2232 true line ending.
2233 """
2234 return self.readuntil(b"\n")
2236 def __next__(self):
2237 out = self.readline()
2238 if out:
2239 return out
2240 raise StopIteration
2242 def __iter__(self):
2243 return self
2245 def readlines(self):
2246 """Return all data, split by the newline character, including the newline character"""
2247 data = self.read()
2248 lines = data.split(b"\n")
2249 out = [l + b"\n" for l in lines[:-1]]
2250 if data.endswith(b"\n"):
2251 return out
2252 else:
2253 return out + [lines[-1]]
2254 # return list(self) ???
2256 def readinto1(self, b):
2257 return self.readinto(b)
2259 def close(self):
2260 """Close file
2262 Finalizes writes, discards cache
2263 """
2264 if getattr(self, "_unclosable", False):
2265 return
2266 if self.closed:
2267 return
2268 try:
2269 if self.mode == "rb":
2270 cache = getattr(self, "cache", None)
2271 if cache is not None:
2272 close = getattr(cache, "close", None)
2273 if callable(close):
2274 close()
2275 self.cache = None
2276 else:
2277 if not getattr(self, "forced", True):
2278 self.flush(force=True)
2280 if self.fs is not None:
2281 self.fs.invalidate_cache(self.path)
2282 self.fs.invalidate_cache(self.fs._parent(self.path))
2283 finally:
2284 self.closed = True
2286 def readable(self):
2287 """Whether opened for reading"""
2288 return "r" in self.mode and not self.closed
2290 def seekable(self):
2291 """Whether is seekable (only in read mode)"""
2292 return self.readable()
2294 def writable(self):
2295 """Whether opened for writing"""
2296 return self.mode in {"wb", "ab", "xb"} and not self.closed
2298 def __reduce__(self):
2299 if self.mode != "rb":
2300 raise RuntimeError("Pickling a writeable file is not supported")
2302 return reopen, (
2303 self.fs,
2304 self.path,
2305 self.mode,
2306 self.blocksize,
2307 self.loc,
2308 self.size,
2309 self.autocommit,
2310 self.cache.name if self.cache else "none",
2311 self.kwargs,
2312 )
2314 def __del__(self):
2315 if not self.closed:
2316 self.close()
2318 def __str__(self):
2319 return f"<File-like object {type(self.fs).__name__}, {self.path}>"
2321 __repr__ = __str__
2323 def __enter__(self):
2324 return self
2326 def __exit__(self, *args):
2327 self.close()
2330def reopen(fs, path, mode, blocksize, loc, size, autocommit, cache_type, kwargs):
2331 file = fs.open(
2332 path,
2333 mode=mode,
2334 block_size=blocksize,
2335 autocommit=autocommit,
2336 cache_type=cache_type,
2337 size=size,
2338 **kwargs,
2339 )
2340 if loc > 0:
2341 file.seek(loc)
2342 return file