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
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 check_contained,
22 glob_translate,
23 isfilelike,
24 other_paths,
25 read_block,
26 stringify_path,
27 tokenize,
28)
30logger = logging.getLogger("fsspec")
33def make_instance(cls, args, kwargs):
34 return cls(*args, **kwargs)
37FORK_AVAILABLE = hasattr(os, "register_at_fork")
40if FORK_AVAILABLE:
41 _registered_classes = weakref.WeakSet()
43 def _reset_instances_lock():
44 for cls in _registered_classes:
45 cls._instantiation_lock = threading.RLock()
46 cls._token_locks = {}
47 cls._cache.clear()
48 cls._pid = os.getpid()
50 os.register_at_fork(after_in_child=_reset_instances_lock)
53class _TokenLock:
54 def __init__(self):
55 self.lock = threading.RLock()
56 self.error = None # set if creating the instance failed
59class _Cached(type):
60 """
61 Metaclass for caching file system instances.
63 Notes
64 -----
65 Instances are cached according to
67 * The values of the class attributes listed in `_extra_tokenize_attributes`
68 * The arguments passed to ``__init__``.
70 This creates an additional reference to the filesystem, which prevents the
71 filesystem from being garbage collected when all *user* references go away.
72 A call to the :meth:`AbstractFileSystem.clear_instance_cache` must *also*
73 be made for a filesystem instance to be garbage collected.
74 """
76 def __init__(cls, *args, **kwargs):
77 super().__init__(*args, **kwargs)
79 # Note: we intentionally create a reference here, to avoid garbage
80 # collecting instances when all other references are gone. To really
81 # delete a FileSystem, the cache must be cleared.
82 if conf.get("weakref_instance_cache"): # pragma: no cover
83 # debug option for analysing fork/spawn conditions
84 cls._cache = weakref.WeakValueDictionary()
85 else:
86 cls._cache = {}
87 cls._pid = os.getpid()
88 cls._instantiation_lock = threading.RLock()
89 # locks of the tokens whose instance is being created, see __call__
90 cls._token_locks = {}
92 if FORK_AVAILABLE:
93 _registered_classes.add(cls)
95 def _check_instance_cache(cls, token):
96 inst = cls._cache.get(token)
97 if inst is not None:
98 cls._latest = token
99 return inst
101 def __call__(cls, *args, **kwargs):
102 kwargs = apply_config(cls, kwargs)
103 extra_tokens = tuple(
104 getattr(cls, attr, None) for attr in cls._extra_tokenize_attributes
105 )
106 strip_tokenize_options = {
107 k: kwargs.pop(k) for k in cls._strip_tokenize_options if k in kwargs
108 }
109 pid = os.getpid()
111 if getattr(cls, "async_impl", False) and not kwargs.get("asynchronous", False):
112 token = tokenize(cls, pid, *args, *extra_tokens, **kwargs)
113 else:
114 token = tokenize(
115 cls, pid, threading.get_ident(), *args, *extra_tokens, **kwargs
116 )
117 skip = kwargs.pop("skip_instance_cache", False)
119 if pid != cls._pid:
120 with cls._instantiation_lock:
121 if pid != cls._pid:
122 cls._cache.clear()
123 cls._token_locks.clear()
124 cls._pid = pid
126 if not skip and cls.cachable:
127 inst = cls._check_instance_cache(token)
128 if inst is not None:
129 return inst
131 with cls._instantiation_lock:
132 # protect against the race condition that a new instance was created
133 # and inserted into the cache since the initial check just above
134 inst = cls._check_instance_cache(token)
135 if inst is not None:
136 return inst
137 # Only one thread creates the instance for a given token, the others
138 # wait for it instead of creating (and discarding) their own, since
139 # creating an instance can be expensive (e.g. looking up credentials).
140 # Instances for different tokens are still created concurrently.
141 # This is under _instantiation_lock, so all threads get the same lock.
142 token_lock = cls._token_locks.setdefault(token, _TokenLock())
144 with token_lock.lock:
145 if token_lock.error is not None:
146 # Fail like the thread we waited for instead of retrying: with
147 # the same arguments it most likely fails the same way, and the
148 # last waiter would fail only after all the others retried.
149 raise token_lock.error
150 inst = cls._check_instance_cache(token)
151 if inst is not None:
152 return inst
153 try:
154 return cls.__create_instance(
155 token, args, kwargs, strip_tokenize_options, cache=True
156 )
157 except Exception as e:
158 token_lock.error = e
159 raise
160 finally:
161 # removed even on failure, so that later calls try again
162 with cls._instantiation_lock:
163 cls._token_locks.pop(token, None)
165 return cls.__create_instance(
166 token, args, kwargs, strip_tokenize_options, cache=False
167 )
169 def __create_instance(cls, token, args, kwargs, strip_tokenize_options, cache):
170 obj = super().__call__(*args, **kwargs, **strip_tokenize_options)
171 # Setting _fs_token here causes some static linters to complain.
172 obj._fs_token_ = token
173 obj.storage_args = args
174 obj.storage_options = kwargs
175 if obj.async_impl and obj.mirror_sync_methods:
176 from .asyn import mirror_sync_methods
178 mirror_sync_methods(obj)
180 if cache:
181 with cls._instantiation_lock:
182 # another thread may have created the instance while we were calling
183 # super().__call__(), so we check again.
184 inst = cls._check_instance_cache(token)
185 if inst is not None:
186 return inst
188 cls._latest = token
189 cls._cache[token] = obj
190 return obj
193class AbstractFileSystem(metaclass=_Cached):
194 """
195 An abstract super-class for pythonic file-systems
197 Implementations are expected to be compatible with or, better, subclass
198 from here.
199 """
201 cachable = True # this class can be cached, instances reused
202 _cached = False
203 blocksize = 2**22
204 sep = "/"
205 # Implementations may select their protocol per instance (for example, a
206 # single adapter class backed by different storage implementations).
207 protocol: str | tuple[str, ...] = "abstract"
208 _latest = None
209 async_impl = False
210 mirror_sync_methods = False
211 root_marker = "" # For some FSs, may require leading '/' or other character
212 transaction_type = Transaction
214 #: Extra *class attributes* that should be considered when hashing.
215 _extra_tokenize_attributes = ()
216 #: *storage options* that should not be considered when hashing.
217 _strip_tokenize_options = ()
219 # Set by _Cached metaclass
220 storage_args: tuple[Any, ...]
221 storage_options: dict[str, Any]
223 def __init__(self, *args, **storage_options):
224 """Create and configure file-system instance
226 Instances may be cachable, so if similar enough arguments are seen
227 a new instance is not required. The token attribute exists to allow
228 implementations to cache instances if they wish.
230 A reasonable default should be provided if there are no arguments.
232 Subclasses should call this method.
234 Parameters
235 ----------
236 use_listings_cache, listings_expiry_time, max_paths:
237 passed to ``DirCache``, if the implementation supports
238 directory listing caching. Pass use_listings_cache=False
239 to disable such caching.
240 skip_instance_cache: bool
241 If this is a cachable implementation, pass True here to force
242 creating a new instance even if a matching instance exists, and prevent
243 storing this instance.
244 asynchronous: bool
245 loop: asyncio-compatible IOLoop or None
246 """
247 if self._cached:
248 # reusing instance, don't change
249 return
250 self._cached = True
251 self._intrans = False
252 self._transaction = None
253 self._invalidated_caches_in_transaction = []
254 self.dircache = DirCache(**storage_options)
256 if storage_options.pop("add_docs", None):
257 warnings.warn("add_docs is no longer supported.", FutureWarning)
259 if storage_options.pop("add_aliases", None):
260 warnings.warn("add_aliases has been removed.", FutureWarning)
261 # This is set in _Cached
262 self._fs_token_ = None
264 @property
265 def fsid(self):
266 """Persistent filesystem id that can be used to compare filesystems
267 across sessions.
268 """
269 raise NotImplementedError
271 @property
272 def _fs_token(self):
273 return self._fs_token_
275 def __dask_tokenize__(self):
276 return self._fs_token
278 def __hash__(self):
279 return int(self._fs_token, 16)
281 def __eq__(self, other):
282 return isinstance(other, type(self)) and self._fs_token == other._fs_token
284 def __reduce__(self):
285 return make_instance, (type(self), self.storage_args, self.storage_options)
287 @classmethod
288 def _strip_protocol(cls, path):
289 """Turn path from fully-qualified to file-system-specific
291 May require FS-specific handling, e.g., for relative paths or links.
292 """
293 if isinstance(path, list):
294 return [cls._strip_protocol(p) for p in path]
295 path = stringify_path(path)
296 protos = (cls.protocol,) if isinstance(cls.protocol, str) else cls.protocol
297 for protocol in protos:
298 if path.startswith(protocol + "://"):
299 path = path[len(protocol) + 3 :]
300 elif path.startswith(protocol + "::"):
301 path = path[len(protocol) + 2 :]
302 path = path.rstrip("/")
303 # use of root_marker to make minimum required path, e.g., "/"
304 return path or cls.root_marker
306 def unstrip_protocol(self, name: str) -> str:
307 """Format FS-specific path to generic, including protocol"""
308 protos = (self.protocol,) if isinstance(self.protocol, str) else self.protocol
309 for protocol in protos:
310 if name.startswith(f"{protocol}://"):
311 return name
312 return f"{protos[0]}://{name}"
314 @staticmethod
315 def _get_kwargs_from_urls(path):
316 """If kwargs can be encoded in the paths, extract them here
318 This should happen before instantiation of the class; incoming paths
319 then should be amended to strip the options in methods.
321 Examples may look like an sftp path "sftp://user@host:/my/path", where
322 the user and host should become kwargs and later get stripped.
323 """
324 # by default, nothing happens
325 return {}
327 @classmethod
328 def current(cls):
329 """Return the most recently instantiated FileSystem
331 If no instance has been created, then create one with defaults
332 """
333 inst = cls._cache.get(cls._latest)
334 if inst is not None:
335 return inst
336 return cls()
338 @property
339 def transaction(self):
340 """A context within which files are committed together upon exit
342 Requires the file class to implement `.commit()` and `.discard()`
343 for the normal and exception cases.
344 """
345 if self._transaction is None:
346 self._transaction = self.transaction_type(self)
347 return self._transaction
349 def start_transaction(self):
350 """Begin write transaction for deferring files, non-context version"""
351 self._intrans = True
352 self._transaction = self.transaction_type(self)
353 return self.transaction
355 def end_transaction(self):
356 """Finish write transaction, non-context version"""
357 self.transaction.complete()
358 self._transaction = None
359 # The invalid cache must be cleared after the transaction is completed.
360 for path in self._invalidated_caches_in_transaction:
361 self.invalidate_cache(path)
362 self._invalidated_caches_in_transaction.clear()
364 def invalidate_cache(self, path=None):
365 """
366 Discard any cached directory information
368 Parameters
369 ----------
370 path: string or None
371 If None, clear all listings cached else listings at or under given
372 path.
373 """
374 # Not necessary to implement invalidation mechanism, may have no cache.
375 # But if have, you should call this method of parent class from your
376 # subclass to ensure expiring caches after transacations correctly.
377 # See the implementation of FTPFileSystem in ftp.py
378 if self._intrans:
379 self._invalidated_caches_in_transaction.append(path)
381 def mkdir(self, path, create_parents=True, **kwargs):
382 """
383 Create directory entry at path
385 For systems that don't have true directories, may create an for
386 this instance only and not touch the real filesystem
388 Parameters
389 ----------
390 path: str
391 location
392 create_parents: bool
393 if True, this is equivalent to ``makedirs``
394 kwargs:
395 may be permissions, etc.
396 """
397 pass # not necessary to implement, may not have directories
399 def makedirs(self, path, exist_ok=False):
400 """Recursively make directories
402 Creates directory at path and any intervening required directories.
403 Raises exception if, for instance, the path already exists but is a
404 file.
406 Parameters
407 ----------
408 path: str
409 leaf directory name
410 exist_ok: bool (False)
411 If False, will error if the target already exists
412 """
413 pass # not necessary to implement, may not have directories
415 def rmdir(self, path):
416 """Remove a directory, if empty"""
417 pass # not necessary to implement, may not have directories
419 def ls(self, path, detail=True, **kwargs):
420 """List objects at path.
422 This should include subdirectories and files at that location. The
423 difference between a file and a directory must be clear when details
424 are requested.
426 The specific keys, or perhaps a FileInfo class, or similar, is TBD,
427 but must be consistent across implementations.
428 Must include:
430 - full path to the entry (without protocol)
431 - size of the entry, in bytes. If the value cannot be determined, will
432 be ``None``.
433 - type of entry, "file", "directory" or other
435 Additional information
436 may be present, appropriate to the file-system, e.g., generation,
437 checksum, etc.
439 May use refresh=True|False to allow use of self._ls_from_cache to
440 check for a saved listing and avoid calling the backend. This would be
441 common where listing may be expensive.
443 Parameters
444 ----------
445 path: str
446 detail: bool
447 if True, gives a list of dictionaries, where each is the same as
448 the result of ``info(path)``. If False, gives a list of paths
449 (str).
450 kwargs: may have additional backend-specific options, such as version
451 information
453 Returns
454 -------
455 List of strings if detail is False, or list of directory information
456 dicts if detail is True.
457 """
458 raise NotImplementedError
460 def _ls_from_cache(self, path):
461 """Check cache for listing
463 Returns listing, if found (may be empty list for a directly that exists
464 but contains nothing), None if not in cache.
465 """
466 parent = self._parent(path)
467 try:
468 return self.dircache[path.rstrip("/")]
469 except KeyError:
470 pass
471 try:
472 files = [
473 f
474 for f in self.dircache[parent]
475 if f["name"] == path
476 or (f["name"] == path.rstrip("/") and f["type"] == "directory")
477 ]
478 if len(files) == 0:
479 # parent dir was listed but did not contain this file
480 raise FileNotFoundError(path)
481 return files
482 except KeyError:
483 pass
485 def walk(self, path, maxdepth=None, topdown=True, on_error="omit", **kwargs):
486 """Return all files under the given path.
488 List all files, recursing into subdirectories; output is iterator-style,
489 like ``os.walk()``. For a simple list of files, ``find()`` is available.
491 When topdown is True, the caller can modify the dirnames list in-place (perhaps
492 using del or slice assignment), and walk() will
493 only recurse into the subdirectories whose names remain in dirnames;
494 this can be used to prune the search, impose a specific order of visiting,
495 or even to inform walk() about directories the caller creates or renames before
496 it resumes walk() again.
497 Modifying dirnames when topdown is False has no effect. (see os.walk)
499 Note that the "files" outputted will include anything that is not
500 a directory, such as links.
502 Parameters
503 ----------
504 path: str
505 Root to recurse into
506 maxdepth: int
507 Maximum recursion depth. None means limitless, but not recommended
508 on link-based file-systems.
509 topdown: bool (True)
510 Whether to walk the directory tree from the top downwards or from
511 the bottom upwards.
512 on_error: "omit", "raise", a callable
513 if omit (default), path with exception will simply be empty;
514 If raise, an underlying exception will be raised;
515 if callable, it will be called with a single OSError instance as argument
516 kwargs: passed to ``ls``
517 """
518 if maxdepth is not None and maxdepth < 1:
519 raise ValueError("maxdepth must be at least 1")
521 path = self._strip_protocol(path)
522 full_dirs = {}
523 dirs = {}
524 files = {}
526 detail = kwargs.pop("detail", False)
527 try:
528 listing = self.ls(path, detail=True, **kwargs)
529 except (FileNotFoundError, OSError) as e:
530 if on_error == "raise":
531 raise
532 if callable(on_error):
533 on_error(e)
534 return
536 for info in listing:
537 # each info name must be at least [path]/part , but here
538 # we check also for names like [path]/part/
539 pathname = info["name"].rstrip("/")
540 name = pathname.rsplit("/", 1)[-1]
541 if info["type"] == "directory" and pathname != path:
542 # do not include "self" path
543 full_dirs[name] = pathname
544 dirs[name] = info
545 elif pathname == path:
546 # file-like with same name as give path
547 files[""] = info
548 else:
549 files[name] = info
551 if not detail:
552 dirs = list(dirs)
553 files = list(files)
555 if topdown:
556 # Yield before recursion if walking top down
557 yield path, dirs, files
559 if maxdepth is not None:
560 maxdepth -= 1
561 if maxdepth < 1:
562 if not topdown:
563 yield path, dirs, files
564 return
566 for d in dirs:
567 yield from self.walk(
568 full_dirs[d],
569 maxdepth=maxdepth,
570 detail=detail,
571 topdown=topdown,
572 on_error=on_error,
573 **kwargs,
574 )
576 if not topdown:
577 # Yield after recursion if walking bottom up
578 yield path, dirs, files
580 def find(self, path, maxdepth=None, withdirs=False, detail=False, **kwargs):
581 """List all files below path.
583 Like posix ``find`` command without conditions
585 Parameters
586 ----------
587 path : str
588 maxdepth: int or None
589 If not None, the maximum number of levels to descend
590 withdirs: bool
591 Whether to include directory paths in the output. This is True
592 when used by glob, but users usually only want files.
593 kwargs are passed to ``ls``.
594 """
595 # TODO: allow equivalent of -name parameter
596 path = self._strip_protocol(path)
597 out = {}
599 # Add the root directory if withdirs is requested
600 # This is needed for posix glob compliance
601 if withdirs and path != "" and self.isdir(path):
602 out[path] = self.info(path)
604 for _, dirs, files in self.walk(path, maxdepth, detail=True, **kwargs):
605 if withdirs:
606 files.update(dirs)
607 out.update({info["name"]: info for name, info in files.items()})
608 if not out and self.isfile(path):
609 # walk works on directories, but find should also return [path]
610 # when path happens to be a file
611 out[path] = {}
612 names = sorted(out)
613 if not detail:
614 return names
615 else:
616 return {name: out[name] for name in names}
618 def du(self, path, total=True, maxdepth=None, withdirs=False, **kwargs):
619 """Space used by files and optionally directories within a path
621 Directory size does not include the size of its contents.
623 Parameters
624 ----------
625 path: str
626 total: bool
627 Whether to sum all the file sizes
628 maxdepth: int or None
629 Maximum number of directory levels to descend, None for unlimited.
630 withdirs: bool
631 Whether to include directory paths in the output.
632 kwargs: passed to ``find``
634 Returns
635 -------
636 Dict of {path: size} if total=False, or int otherwise, where numbers
637 refer to bytes used.
638 """
639 sizes = {}
640 if withdirs and self.isdir(path):
641 # Include top-level directory in output
642 info = self.info(path)
643 sizes[info["name"]] = info["size"]
644 for f in self.find(path, maxdepth=maxdepth, withdirs=withdirs, **kwargs):
645 info = self.info(f)
646 sizes[info["name"]] = info["size"]
647 if total:
648 return sum(sizes.values())
649 else:
650 return sizes
652 def glob(self, path, maxdepth=None, **kwargs):
653 """Find files by glob-matching.
655 Pattern matching capabilities for finding files that match the given pattern.
657 Parameters
658 ----------
659 path: str
660 The glob pattern to match against
661 maxdepth: int or None
662 Maximum depth for ``'**'`` patterns. Applied on the first ``'**'`` found.
663 Must be at least 1 if provided.
664 kwargs:
665 Additional arguments passed to ``find`` (e.g., detail=True)
667 Returns
668 -------
669 List of matched paths, or dict of paths and their info if detail=True
671 Notes
672 -----
673 Supported patterns:
674 - '*': Matches any sequence of characters within a single directory level
675 - ``'**'``: Matches any number of directory levels (must be an entire path component)
676 - '?': Matches exactly one character
677 - '[abc]': Matches any character in the set
678 - '[a-z]': Matches any character in the range
679 - '[!abc]': Matches any character NOT in the set
681 Special behaviors:
682 - If the path ends with '/', only folders are returned
683 - Consecutive '*' characters are compressed into a single '*'
684 - Empty set '[]' or negated empty negated set '[!]' never match anything
685 - Special characters in character classes are escaped properly
687 Limitations:
688 - ``'**'`` must be a complete path component (e.g., ``'a/**/b'``, not ``'a**b'``)
689 - No brace expansion ('{a,b}.txt')
690 - No extended glob patterns ('+(pattern)', '!(pattern)')
691 """
692 if maxdepth is not None and maxdepth < 1:
693 raise ValueError("maxdepth must be at least 1")
695 import re
697 seps = (os.path.sep, os.path.altsep) if os.path.altsep else (os.path.sep,)
698 ends_with_sep = path.endswith(seps) # _strip_protocol strips trailing slash
699 path = self._strip_protocol(path)
700 append_slash_to_dirname = ends_with_sep or path.endswith(
701 tuple(sep + "**" for sep in seps)
702 )
703 idx_star = path.find("*") if path.find("*") >= 0 else len(path)
704 idx_qmark = path.find("?") if path.find("?") >= 0 else len(path)
705 idx_brace = path.find("[") if path.find("[") >= 0 else len(path)
707 min_idx = min(idx_star, idx_qmark, idx_brace)
709 detail = kwargs.pop("detail", False)
710 withdirs = kwargs.pop("withdirs", True)
712 if not has_magic(path):
713 if self.exists(path, **kwargs):
714 if not detail:
715 return [path]
716 else:
717 return {path: self.info(path, **kwargs)}
718 else:
719 if not detail:
720 return [] # glob of non-existent returns empty
721 else:
722 return {}
723 elif "/" in path[:min_idx]:
724 min_idx = path[:min_idx].rindex("/")
725 root = path[: min_idx + 1]
726 depth = path[min_idx + 1 :].count("/") + 1
727 else:
728 root = ""
729 depth = path[min_idx + 1 :].count("/") + 1
731 if "**" in path:
732 if maxdepth is not None:
733 idx_double_stars = path.find("**")
734 depth_double_stars = path[idx_double_stars:].count("/") + 1
735 depth = depth - depth_double_stars + maxdepth
736 else:
737 depth = None
739 allpaths = self.find(
740 root, maxdepth=depth, withdirs=withdirs, detail=True, **kwargs
741 )
743 pattern = glob_translate(path + ("/" if ends_with_sep else ""))
744 pattern = re.compile(pattern)
746 out = {
747 p: info
748 for p, info in sorted(allpaths.items())
749 if pattern.match(
750 p + "/"
751 if append_slash_to_dirname and info["type"] == "directory"
752 else p
753 )
754 }
756 if detail:
757 return out
758 else:
759 return list(out)
761 def exists(self, path, **kwargs):
762 """Is there a file at the given path"""
763 try:
764 self.info(path, **kwargs)
765 return True
766 except: # noqa: E722
767 # any exception allowed bar FileNotFoundError?
768 return False
770 def lexists(self, path, **kwargs):
771 """If there is a file at the given path (including
772 broken links)"""
773 return self.exists(path)
775 def info(self, path, **kwargs):
776 """Give details of entry at path
778 Returns a single dictionary, with exactly the same information as ``ls``
779 would with ``detail=True``.
781 The default implementation calls ls and could be overridden by a
782 shortcut. kwargs are passed on to ```ls()``.
784 Some file systems might not be able to measure the file's size, in
785 which case, the returned dict will include ``'size': None``.
787 Returns
788 -------
789 dict with keys: name (full path in the FS), size (in bytes), type (file,
790 directory, or something else) and other FS-specific keys.
791 """
792 path = self._strip_protocol(path)
793 out = self.ls(self._parent(path), detail=True, **kwargs)
794 out = [o for o in out if o["name"].rstrip("/") == path]
795 if out:
796 return out[0]
797 out = self.ls(path, detail=True, **kwargs)
798 path = path.rstrip("/")
799 out1 = [o for o in out if o["name"].rstrip("/") == path]
800 if len(out1) == 1:
801 if "size" not in out1[0]:
802 out1[0]["size"] = None
803 return out1[0]
804 elif len(out1) > 1 or out:
805 return {"name": path, "size": 0, "type": "directory"}
806 else:
807 raise FileNotFoundError(path)
809 def checksum(self, path):
810 """Unique value for current version of file
812 If the checksum is the same from one moment to another, the contents
813 are guaranteed to be the same. If the checksum changes, the contents
814 *might* have changed.
816 This should normally be overridden; default will probably capture
817 creation/modification timestamp (which would be good) or maybe
818 access timestamp (which would be bad)
819 """
820 return int(tokenize(self.info(path)), 16)
822 def size(self, path):
823 """Size in bytes of file"""
824 return self.info(path).get("size", None)
826 def sizes(self, paths):
827 """Size in bytes of each file in a list of paths"""
828 return [self.size(p) for p in paths]
830 def isdir(self, path):
831 """Is this entry directory-like?"""
832 try:
833 return self.info(path)["type"] == "directory"
834 except OSError:
835 return False
837 def isfile(self, path):
838 """Is this entry file-like?"""
839 try:
840 return self.info(path)["type"] == "file"
841 except Exception:
842 return False
844 def read_text(self, path, encoding=None, errors=None, newline=None, **kwargs):
845 """Get the contents of the file as a string.
847 Parameters
848 ----------
849 path: str
850 URL of file on this filesystems
851 encoding, errors, newline: same as `open`.
852 """
853 with self.open(
854 path,
855 mode="r",
856 encoding=encoding,
857 errors=errors,
858 newline=newline,
859 **kwargs,
860 ) as f:
861 return f.read()
863 def write_text(
864 self, path, value, encoding=None, errors=None, newline=None, **kwargs
865 ):
866 """Write the text to the given file.
868 An existing file will be overwritten.
870 Parameters
871 ----------
872 path: str
873 URL of file on this filesystems
874 value: str
875 Text to write.
876 encoding, errors, newline: same as `open`.
877 """
878 with self.open(
879 path,
880 mode="w",
881 encoding=encoding,
882 errors=errors,
883 newline=newline,
884 **kwargs,
885 ) as f:
886 return f.write(value)
888 def cat_file(self, path, start=None, end=None, **kwargs):
889 """Get the content of a file
891 Parameters
892 ----------
893 path: URL of file on this filesystems
894 start, end: int
895 Bytes limits of the read. If negative, backwards from end,
896 like usual python slices. Either can be None for start or
897 end of file, respectively
898 kwargs: passed to ``open()``.
899 """
900 # explicitly set buffering off?
901 with self.open(path, "rb", **kwargs) as f:
902 # A negative start or end counts back from the file length. Only
903 # fsspec's own file classes expose `.size`; a plain file (e.g. from
904 # the caching filesystems) does not, so measure with a seek instead.
905 needs_size = (start is not None and start < 0) or (
906 end is not None and end < 0
907 )
908 size = None
909 seeked_for_size = False
910 if needs_size:
911 size = getattr(f, "size", None)
912 if size is None:
913 size = f.seek(0, 2)
914 seeked_for_size = True
915 if start is not None:
916 if start >= 0:
917 f.seek(start)
918 else:
919 f.seek(max(0, size + start))
920 elif seeked_for_size:
921 # measuring the size above moved the cursor to the end; with no
922 # start given, put it back so the read begins at the start. When
923 # .size was used instead, or nothing was measured, the cursor
924 # is untouched, as before, so a plain full read still works
925 # on non-seekable files.
926 f.seek(0)
927 if end is not None:
928 if end < 0:
929 # a negative end counts back from the end, as the start
930 # does, and clamps at the start of the file
931 end = max(0, size + end)
932 # a range that ends at or before where it starts is empty; a
933 # negative length would instead mean "read to the end"
934 return f.read(max(0, end - f.tell()))
935 return f.read()
937 def pipe_file(self, path, value, mode="overwrite", **kwargs):
938 """Set the bytes of given file"""
939 if mode == "create" and self.exists(path):
940 # non-atomic but simple way; or could use "xb" in open(), which is likely
941 # not as well supported
942 raise FileExistsError
943 with self.open(path, "wb", **kwargs) as f:
944 f.write(value)
946 def pipe(self, path, value=None, **kwargs):
947 """Put value into path
949 (counterpart to ``cat``)
951 Parameters
952 ----------
953 path: string or dict(str, bytes)
954 If a string, a single remote location to put ``value`` bytes; if a dict,
955 a mapping of {path: bytesvalue}.
956 value: bytes, optional
957 If using a single path, these are the bytes to put there. Ignored if
958 ``path`` is a dict
959 """
960 if isinstance(path, str):
961 self.pipe_file(self._strip_protocol(path), value, **kwargs)
962 elif isinstance(path, dict):
963 for k, v in path.items():
964 self.pipe_file(self._strip_protocol(k), v, **kwargs)
965 else:
966 raise ValueError("path must be str or dict")
968 def cat_ranges(
969 self, paths, starts, ends, max_gap=None, on_error="return", **kwargs
970 ):
971 """Get the contents of byte ranges from one or more files
973 Parameters
974 ----------
975 paths: list
976 A list of of filepaths on this filesystems
977 starts, ends: int or list
978 Bytes limits of the read. If using a single int, the same value will be
979 used to read all the specified files.
980 """
981 if max_gap is not None:
982 raise NotImplementedError
983 if not isinstance(paths, list):
984 raise TypeError
985 if not isinstance(starts, list):
986 starts = [starts] * len(paths)
987 if not isinstance(ends, list):
988 ends = [ends] * len(paths)
989 if len(starts) != len(paths) or len(ends) != len(paths):
990 raise ValueError
991 out = []
992 for p, s, e in zip(paths, starts, ends):
993 try:
994 out.append(self.cat_file(p, s, e, **kwargs))
995 except Exception as e:
996 if on_error == "return":
997 out.append(e)
998 else:
999 raise
1000 return out
1002 def cat(self, path, recursive=False, on_error="raise", **kwargs):
1003 """Fetch (potentially multiple) paths' contents
1005 Parameters
1006 ----------
1007 recursive: bool
1008 If True, assume the path(s) are directories, and get all the
1009 contained files
1010 on_error : "raise", "omit", "return"
1011 If raise, an underlying exception will be raised (converted to KeyError
1012 if the type is in self.missing_exceptions); if omit, keys with exception
1013 will simply not be included in the output; if "return", all keys are
1014 included in the output, but the value will be bytes or an exception
1015 instance.
1016 kwargs: passed to cat_file
1018 Returns
1019 -------
1020 dict of {path: contents} if there are multiple paths
1021 or the path has been otherwise expanded
1022 """
1023 paths = self.expand_path(path, recursive=recursive, **kwargs)
1024 if (
1025 len(paths) > 1
1026 or isinstance(path, list)
1027 or paths[0] != self._strip_protocol(path)
1028 ):
1029 out = {}
1030 for path in paths:
1031 try:
1032 out[path] = self.cat_file(path, **kwargs)
1033 except Exception as e:
1034 if on_error == "raise":
1035 raise
1036 if on_error == "return":
1037 out[path] = e
1038 return out
1039 else:
1040 return self.cat_file(paths[0], **kwargs)
1042 def get_file(
1043 self, rpath, lpath=None, callback=DEFAULT_CALLBACK, outfile=None, **kwargs
1044 ):
1045 """Copy single remote file to local"""
1046 from .implementations.local import LocalFileSystem
1048 if outfile is None and isfilelike(lpath):
1049 outfile = lpath
1050 elif outfile is None and self.isdir(rpath):
1051 os.makedirs(lpath, exist_ok=True)
1052 return None
1054 if outfile is None:
1055 fs = LocalFileSystem(auto_mkdir=True)
1056 fs.makedirs(fs._parent(lpath), exist_ok=True)
1058 with self.open(rpath, "rb", **kwargs) as f1:
1059 close_outfile = outfile is None
1060 if close_outfile:
1061 outfile = open(lpath, "wb")
1063 try:
1064 callback.set_size(getattr(f1, "size", None))
1065 data = True
1066 while data:
1067 data = f1.read(self.blocksize)
1068 segment_len = outfile.write(data)
1069 if segment_len is None:
1070 segment_len = len(data)
1071 callback.relative_update(segment_len)
1072 finally:
1073 if close_outfile:
1074 outfile.close()
1076 def get(
1077 self,
1078 rpath,
1079 lpath,
1080 recursive=False,
1081 callback=DEFAULT_CALLBACK,
1082 maxdepth=None,
1083 **kwargs,
1084 ):
1085 """Copy file(s) to local.
1087 Copies a specific file or tree of files (if recursive=True). If lpath
1088 ends with a "/", it will be assumed to be a directory, and target files
1089 will go within. Can submit a list of paths, which may be glob-patterns
1090 and will be expanded.
1092 Calls get_file for each source.
1093 """
1094 if isinstance(lpath, list) and isinstance(rpath, list):
1095 # No need to expand paths when both source and destination
1096 # are provided as lists
1097 rpaths = rpath
1098 lpaths = lpath
1099 else:
1100 from .implementations.local import (
1101 LocalFileSystem,
1102 make_path_posix,
1103 trailing_sep,
1104 )
1106 rpath = stringify_path(rpath)
1107 source_is_str = isinstance(rpath, str)
1108 rpaths = self.expand_path(
1109 rpath, recursive=recursive, maxdepth=maxdepth, **kwargs
1110 )
1111 if source_is_str and (not recursive or maxdepth is not None):
1112 # Non-recursive glob does not copy directories
1113 rpaths = [p for p in rpaths if not (trailing_sep(p) or self.isdir(p))]
1114 if not rpaths:
1115 return
1117 if isinstance(lpath, (str, os.PathLike)):
1118 lpath = make_path_posix(lpath)
1120 source_is_file = len(rpaths) == 1
1121 dest_is_dir = isinstance(lpath, str) and (
1122 trailing_sep(lpath) or LocalFileSystem().isdir(lpath)
1123 )
1125 exists = source_is_str and (
1126 (has_magic(rpath) and source_is_file)
1127 or (not has_magic(rpath) and dest_is_dir and not trailing_sep(rpath))
1128 )
1129 lpaths = other_paths(
1130 rpaths,
1131 lpath,
1132 exists=exists,
1133 flatten=not source_is_str,
1134 )
1135 if isinstance(lpath, str):
1136 # The names came from the source listing; ".." in one of them
1137 # would otherwise place the copy above the destination. When
1138 # lpath is a list the caller named every destination itself.
1139 check_contained(lpath, lpaths)
1141 callback.set_size(len(lpaths))
1142 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):
1143 with callback.branched(rpath, lpath) as child:
1144 self.get_file(rpath, lpath, callback=child, **kwargs)
1146 def put_file(
1147 self, lpath, rpath, callback=DEFAULT_CALLBACK, mode="overwrite", **kwargs
1148 ):
1149 """Copy single file to remote"""
1150 if mode == "create" and self.exists(rpath):
1151 raise FileExistsError
1152 if os.path.isdir(lpath):
1153 self.makedirs(rpath, exist_ok=True)
1154 return None
1156 with open(lpath, "rb") as f1:
1157 size = f1.seek(0, 2)
1158 callback.set_size(size)
1159 f1.seek(0)
1161 self.mkdirs(self._parent(os.fspath(rpath)), exist_ok=True)
1162 with self.open(rpath, "wb", **kwargs) as f2:
1163 while f1.tell() < size:
1164 data = f1.read(self.blocksize)
1165 segment_len = f2.write(data)
1166 if segment_len is None:
1167 segment_len = len(data)
1168 callback.relative_update(segment_len)
1170 def put(
1171 self,
1172 lpath,
1173 rpath,
1174 recursive=False,
1175 callback=DEFAULT_CALLBACK,
1176 maxdepth=None,
1177 **kwargs,
1178 ):
1179 """Copy file(s) from local.
1181 Copies a specific file or tree of files (if recursive=True). If rpath
1182 ends with a "/", it will be assumed to be a directory, and target files
1183 will go within.
1185 Calls put_file for each source.
1186 """
1187 if isinstance(lpath, list) and isinstance(rpath, list):
1188 # No need to expand paths when both source and destination
1189 # are provided as lists
1190 rpaths = rpath
1191 lpaths = lpath
1192 else:
1193 from .implementations.local import (
1194 LocalFileSystem,
1195 make_path_posix,
1196 trailing_sep,
1197 )
1199 lpath = stringify_path(lpath)
1200 source_is_str = isinstance(lpath, str)
1201 if source_is_str:
1202 lpath = make_path_posix(lpath)
1203 fs = LocalFileSystem()
1204 lpaths = fs.expand_path(
1205 lpath, recursive=recursive, maxdepth=maxdepth, **kwargs
1206 )
1207 if source_is_str and (not recursive or maxdepth is not None):
1208 # Non-recursive glob does not copy directories
1209 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
1210 if not lpaths:
1211 return
1213 source_is_file = len(lpaths) == 1
1214 dest_is_dir = isinstance(rpath, str) and (
1215 trailing_sep(rpath) or self.isdir(rpath)
1216 )
1218 rpath = (
1219 self._strip_protocol(rpath)
1220 if isinstance(rpath, str)
1221 else [self._strip_protocol(p) for p in rpath]
1222 )
1223 exists = source_is_str and (
1224 (has_magic(lpath) and source_is_file)
1225 or (not has_magic(lpath) and dest_is_dir and not trailing_sep(lpath))
1226 )
1227 rpaths = other_paths(
1228 lpaths,
1229 rpath,
1230 exists=exists,
1231 flatten=not source_is_str,
1232 )
1234 callback.set_size(len(rpaths))
1235 for lpath, rpath in callback.wrap(zip(lpaths, rpaths)):
1236 with callback.branched(lpath, rpath) as child:
1237 self.put_file(lpath, rpath, callback=child, **kwargs)
1239 def head(self, path, size=1024):
1240 """Get the first ``size`` bytes from file"""
1241 with self.open(path, "rb") as f:
1242 return f.read(size)
1244 def tail(self, path, size=1024):
1245 """Get the last ``size`` bytes from file"""
1246 with self.open(path, "rb") as f:
1247 f.seek(0, 2)
1248 f.seek(max(f.tell() - size, 0))
1249 return f.read()
1251 def cp_file(self, path1, path2, **kwargs):
1252 raise NotImplementedError
1254 def copy(
1255 self, path1, path2, recursive=False, maxdepth=None, on_error=None, **kwargs
1256 ):
1257 """Copy within two locations in the filesystem
1259 on_error : "raise", "ignore"
1260 If raise, any not-found exceptions will be raised; if ignore any
1261 not-found exceptions will cause the path to be skipped; defaults to
1262 raise unless recursive is true, where the default is ignore
1263 """
1264 if on_error is None and recursive:
1265 on_error = "ignore"
1266 elif on_error is None:
1267 on_error = "raise"
1269 if isinstance(path1, list) and isinstance(path2, list):
1270 # No need to expand paths when both source and destination
1271 # are provided as lists
1272 paths1 = path1
1273 paths2 = path2
1274 else:
1275 from .implementations.local import trailing_sep
1277 path1 = stringify_path(path1)
1278 source_is_str = isinstance(path1, str)
1279 paths1 = self.expand_path(
1280 path1, recursive=recursive, maxdepth=maxdepth, **kwargs
1281 )
1282 if source_is_str and (not recursive or maxdepth is not None):
1283 # Non-recursive glob does not copy directories
1284 paths1 = [p for p in paths1 if not (trailing_sep(p) or self.isdir(p))]
1285 if not paths1:
1286 return
1288 source_is_file = len(paths1) == 1
1289 dest_is_dir = isinstance(path2, str) and (
1290 trailing_sep(path2) or self.isdir(path2)
1291 )
1293 exists = source_is_str and (
1294 (has_magic(path1) and source_is_file)
1295 or (not has_magic(path1) and dest_is_dir and not trailing_sep(path1))
1296 )
1297 paths2 = other_paths(
1298 paths1,
1299 path2,
1300 exists=exists,
1301 flatten=not source_is_str,
1302 )
1304 for p1, p2 in zip(paths1, paths2):
1305 try:
1306 self.cp_file(p1, p2, **kwargs)
1307 except FileNotFoundError:
1308 if on_error == "raise":
1309 raise
1311 def expand_path(
1312 self, path, recursive=False, maxdepth=None, assume_literal=False, **kwargs
1313 ):
1314 """Turn one or more globs or directories into a list of all matching paths
1315 to files or directories.
1317 kwargs are passed to ``glob`` or ``find``, which may in turn call ``ls``
1318 """
1320 if maxdepth is not None and maxdepth < 1:
1321 raise ValueError("maxdepth must be at least 1")
1323 if isinstance(path, (str, os.PathLike)):
1324 out = self.expand_path([path], recursive, maxdepth, **kwargs)
1325 else:
1326 out = set()
1327 path = [self._strip_protocol(p) for p in path]
1328 for p in path:
1329 if not assume_literal and has_magic(p):
1330 bit = set(self.glob(p, maxdepth=maxdepth, **kwargs))
1331 out |= bit
1332 if recursive:
1333 # glob call above expanded one depth so if maxdepth is defined
1334 # then decrement it in expand_path call below. If it is zero
1335 # after decrementing then avoid expand_path call.
1336 if maxdepth is not None and maxdepth <= 1:
1337 continue
1338 out |= set(
1339 self.expand_path(
1340 list(bit),
1341 recursive=recursive,
1342 maxdepth=maxdepth - 1 if maxdepth is not None else None,
1343 assume_literal=True,
1344 **kwargs,
1345 )
1346 )
1347 continue
1348 elif recursive:
1349 rec = set(
1350 self.find(
1351 p, maxdepth=maxdepth, withdirs=True, detail=False, **kwargs
1352 )
1353 )
1354 out |= rec
1355 if p not in out and (recursive is False or self.exists(p)):
1356 # should only check once, for the root
1357 out.add(p)
1358 if not out:
1359 raise FileNotFoundError(path)
1360 return sorted(out)
1362 def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs):
1363 """Move file(s) from one location to another"""
1364 if path1 == path2:
1365 logger.debug("%s mv: The paths are the same, so no files were moved.", self)
1366 else:
1367 if recursive and maxdepth is not None:
1368 # copy only reaches maxdepth, so only remove what it copied
1369 paths = self.expand_path(path1, recursive=True, maxdepth=maxdepth)
1370 dirs = [p for p in paths if self.isdir(p)]
1371 files = [p for p in paths if p not in dirs]
1372 # explicitly raise exception to prevent data corruption
1373 self.copy(
1374 path1, path2, recursive=recursive, maxdepth=maxdepth, on_error="raise"
1375 )
1376 if recursive and maxdepth is not None:
1377 if files:
1378 self.rm(files)
1379 # a directory still holding files deeper than maxdepth is kept
1380 for p in reversed(dirs):
1381 if self.exists(p) and not self.ls(p, detail=False):
1382 self.rmdir(p)
1383 else:
1384 self.rm(path1, recursive=recursive)
1386 def rm_file(self, path):
1387 """Delete a file"""
1388 self._rm(path)
1390 def _rm(self, path):
1391 """Delete one file"""
1392 # this is the old name for the method, prefer rm_file
1393 raise NotImplementedError
1395 def rm(self, path, recursive=False, maxdepth=None):
1396 """Delete files or directories.
1398 Parameters
1399 ----------
1400 path: str or list of str
1401 Files or directories to delete.
1402 recursive: bool
1403 If True, recursively delete directories and their contents.
1404 maxdepth: int or None
1405 Depth to pass to walk for finding files to delete, if recursive.
1406 If None, there will be no limit and infinite recursion may be
1407 possible.
1408 """
1409 path = self.expand_path(path, recursive=recursive, maxdepth=maxdepth)
1410 for p in reversed(path):
1411 self.rm_file(p)
1413 @classmethod
1414 def _parent(cls, path):
1415 path = cls._strip_protocol(path)
1416 if "/" in path:
1417 parent = path.rsplit("/", 1)[0].lstrip(cls.root_marker)
1418 return cls.root_marker + parent
1419 else:
1420 return cls.root_marker
1422 def _open(
1423 self,
1424 path,
1425 mode="rb",
1426 block_size=None,
1427 autocommit=True,
1428 cache_options=None,
1429 **kwargs,
1430 ):
1431 """Return raw bytes-mode file-like from the file-system"""
1432 return AbstractBufferedFile(
1433 self,
1434 path,
1435 mode,
1436 block_size,
1437 autocommit,
1438 cache_options=cache_options,
1439 **kwargs,
1440 )
1442 def open(
1443 self,
1444 path,
1445 mode="rb",
1446 block_size=None,
1447 cache_options=None,
1448 compression=None,
1449 **kwargs,
1450 ):
1451 """
1452 Return a file-like object from the filesystem
1454 The resultant instance must function correctly in a context ``with``
1455 block.
1457 Parameters
1458 ----------
1459 path: str
1460 Target file
1461 mode: str like 'rb', 'w'
1462 See builtin ``open()``
1463 Mode "x" (exclusive write) may be implemented by the backend. Even if
1464 it is, whether it is checked up front or on commit, and whether it is
1465 atomic is implementation-dependent.
1466 block_size: int
1467 Some indication of buffering - this is a value in bytes
1468 cache_options : dict, optional
1469 Extra arguments to pass through to the cache.
1470 compression: string or None
1471 If given, open file using compression codec. Can either be a compression
1472 name (a key in ``fsspec.compression.compr``) or "infer" to guess the
1473 compression from the filename suffix.
1474 encoding, errors, newline: passed on to TextIOWrapper for text mode
1475 """
1476 import io
1478 path = self._strip_protocol(path)
1479 if "b" not in mode:
1480 mode = mode.replace("t", "") + "b"
1482 text_kwargs = {
1483 k: kwargs.pop(k)
1484 for k in ["encoding", "errors", "newline"]
1485 if k in kwargs
1486 }
1487 return io.TextIOWrapper(
1488 self.open(
1489 path,
1490 mode,
1491 block_size=block_size,
1492 cache_options=cache_options,
1493 compression=compression,
1494 **kwargs,
1495 ),
1496 **text_kwargs,
1497 )
1498 else:
1499 ac = kwargs.pop("autocommit", not self._intrans)
1500 f = self._open(
1501 path,
1502 mode=mode,
1503 block_size=block_size,
1504 autocommit=ac,
1505 cache_options=cache_options,
1506 **kwargs,
1507 )
1508 if not ac and "r" not in mode:
1509 self.transaction.files.append(f)
1510 if compression is not None:
1511 from fsspec.compression import compr
1512 from fsspec.core import get_compression
1514 compression = get_compression(path, compression)
1515 compress = compr[compression]
1516 f = compress(f, mode=mode[0])
1517 return f
1519 def touch(self, path, truncate=True, **kwargs):
1520 """Create empty file, or update timestamp
1522 Parameters
1523 ----------
1524 path: str
1525 file location
1526 truncate: bool
1527 If True, always set file size to 0; if False, update timestamp and
1528 leave file unchanged, if backend allows this
1529 """
1530 if truncate or not self.exists(path):
1531 with self.open(path, "wb", **kwargs):
1532 pass
1533 else:
1534 raise NotImplementedError # update timestamp, if possible
1536 def ukey(self, path):
1537 """Hash of file properties, to tell if it has changed"""
1538 return sha256(str(self.info(path)).encode()).hexdigest()
1540 def read_block(self, fn, offset, length, delimiter=None):
1541 """Read a block of bytes from
1543 Starting at ``offset`` of the file, read ``length`` bytes. If
1544 ``delimiter`` is set then we ensure that the read starts and stops at
1545 delimiter boundaries that follow the locations ``offset`` and ``offset
1546 + length``. If ``offset`` is zero then we start at zero. The
1547 bytestring returned WILL include the end delimiter string.
1549 If offset+length is beyond the eof, reads to eof.
1551 Parameters
1552 ----------
1553 fn: string
1554 Path to filename
1555 offset: int
1556 Byte offset to start read
1557 length: int
1558 Number of bytes to read. If None, read to end.
1559 delimiter: bytes (optional)
1560 Ensure reading starts and stops at delimiter bytestring
1562 Examples
1563 --------
1564 >>> fs.read_block('data/file.csv', 0, 13) # doctest: +SKIP
1565 b'Alice, 100\\nBo'
1566 >>> fs.read_block('data/file.csv', 0, 13, delimiter=b'\\n') # doctest: +SKIP
1567 b'Alice, 100\\nBob, 200\\n'
1569 Use ``length=None`` to read to the end of the file.
1570 >>> fs.read_block('data/file.csv', 0, None, delimiter=b'\\n') # doctest: +SKIP
1571 b'Alice, 100\\nBob, 200\\nCharlie, 300'
1573 See Also
1574 --------
1575 :func:`fsspec.utils.read_block`
1576 """
1577 with self.open(fn, "rb") as f:
1578 # Only fsspec's own file classes expose `.size`; a plain file (e.g.
1579 # from the caching filesystems) does not, so fall back to a seek.
1580 size = getattr(f, "size", None)
1581 if size is None:
1582 size = f.seek(0, 2)
1583 f.seek(0)
1584 if length is None:
1585 length = size
1586 if size is not None and offset + length > size:
1587 length = max(0, size - offset)
1588 return read_block(f, offset, length, delimiter)
1590 def to_json(self, *, include_password: bool = True) -> str:
1591 """
1592 JSON representation of this filesystem instance.
1594 Parameters
1595 ----------
1596 include_password: bool, default True
1597 Whether to include the password (if any) in the output.
1599 Returns
1600 -------
1601 JSON string with keys ``cls`` (the python location of this class),
1602 protocol (text name of this class's protocol, first one in case of
1603 multiple), ``args`` (positional args, usually empty), and all other
1604 keyword arguments as their own keys.
1606 Warnings
1607 --------
1608 Serialized filesystems may contain sensitive information which have been
1609 passed to the constructor, such as passwords and tokens. Make sure you
1610 store and send them in a secure environment!
1611 """
1612 from .json import FilesystemJSONEncoder
1614 return json.dumps(
1615 self,
1616 cls=type(
1617 "_FilesystemJSONEncoder",
1618 (FilesystemJSONEncoder,),
1619 {"include_password": include_password},
1620 ),
1621 )
1623 @staticmethod
1624 def from_json(blob: str) -> AbstractFileSystem:
1625 """
1626 Recreate a filesystem instance from JSON representation.
1628 See ``.to_json()`` for the expected structure of the input.
1630 Parameters
1631 ----------
1632 blob: str
1634 Returns
1635 -------
1636 file system instance, not necessarily of this particular class.
1638 Warnings
1639 --------
1640 This can import arbitrary modules (as determined by the ``cls`` key).
1641 Make sure you haven't installed any modules that may execute malicious code
1642 at import time.
1643 """
1644 from .json import FilesystemJSONDecoder
1646 return json.loads(blob, cls=FilesystemJSONDecoder)
1648 def to_dict(self, *, include_password: bool = True) -> dict[str, Any]:
1649 """
1650 JSON-serializable dictionary representation of this filesystem instance.
1652 Parameters
1653 ----------
1654 include_password: bool, default True
1655 Whether to include the password (if any) in the output.
1657 Returns
1658 -------
1659 Dictionary with keys ``cls`` (the python location of this class),
1660 protocol (text name of this class's protocol, first one in case of
1661 multiple), ``args`` (positional args, usually empty), and all other
1662 keyword arguments as their own keys.
1664 Warnings
1665 --------
1666 Serialized filesystems may contain sensitive information which have been
1667 passed to the constructor, such as passwords and tokens. Make sure you
1668 store and send them in a secure environment!
1669 """
1670 from .json import FilesystemJSONEncoder
1672 json_encoder = type(
1673 "_FilesystemJSONEncoder",
1674 (FilesystemJSONEncoder,),
1675 {"include_password": include_password},
1676 )()
1678 cls = type(self)
1679 proto = self.protocol
1681 storage_options = dict(self.storage_options)
1682 if not include_password:
1683 storage_options.pop("password", None)
1685 return dict(
1686 cls=f"{cls.__module__}:{cls.__name__}",
1687 protocol=proto[0] if isinstance(proto, (tuple, list)) else proto,
1688 args=json_encoder.make_serializable(self.storage_args),
1689 **json_encoder.make_serializable(storage_options),
1690 )
1692 @staticmethod
1693 def from_dict(dct: dict[str, Any]) -> AbstractFileSystem:
1694 """
1695 Recreate a filesystem instance from dictionary representation.
1697 See ``.to_dict()`` for the expected structure of the input.
1699 Parameters
1700 ----------
1701 dct: Dict[str, Any]
1703 Returns
1704 -------
1705 file system instance, not necessarily of this particular class.
1707 Warnings
1708 --------
1709 This can import arbitrary modules (as determined by the ``cls`` key).
1710 Make sure you haven't installed any modules that may execute malicious code
1711 at import time.
1712 """
1713 from .json import FilesystemJSONDecoder
1715 json_decoder = FilesystemJSONDecoder()
1717 dct = dict(dct) # Defensive copy
1719 cls = FilesystemJSONDecoder.try_resolve_fs_cls(dct)
1720 if cls is None:
1721 raise ValueError("Not a serialized AbstractFileSystem")
1723 dct.pop("cls", None)
1724 dct.pop("protocol", None)
1726 return cls(
1727 *json_decoder.unmake_serializable(dct.pop("args", ())),
1728 **json_decoder.unmake_serializable(dct),
1729 )
1731 def _get_pyarrow_filesystem(self):
1732 """
1733 Make a version of the FS instance which will be acceptable to pyarrow
1734 """
1735 # all instances already also derive from pyarrow
1736 return self
1738 def get_mapper(self, root="", check=False, create=False, missing_exceptions=None):
1739 """Create key/value store based on this file-system
1741 Makes a MutableMapping interface to the FS at the given root path.
1742 See ``fsspec.mapping.FSMap`` for further details.
1743 """
1744 from .mapping import FSMap
1746 return FSMap(
1747 root,
1748 self,
1749 check=check,
1750 create=create,
1751 missing_exceptions=missing_exceptions,
1752 )
1754 @classmethod
1755 def clear_instance_cache(cls):
1756 """
1757 Clear the cache of filesystem instances.
1759 Notes
1760 -----
1761 Unless overridden by setting the ``cachable`` class attribute to False,
1762 the filesystem class stores a reference to newly created instances. This
1763 prevents Python's normal rules around garbage collection from working,
1764 since the instances refcount will not drop to zero until
1765 ``clear_instance_cache`` is called.
1766 """
1767 cls._cache.clear()
1769 def created(self, path):
1770 """Return the created timestamp of a file as a datetime.datetime"""
1771 raise NotImplementedError
1773 def modified(self, path):
1774 """Return the modified timestamp of a file as a datetime.datetime"""
1775 raise NotImplementedError
1777 def tree(
1778 self,
1779 path: str = "/",
1780 recursion_limit: int = 2,
1781 max_display: int = 25,
1782 display_size: bool = False,
1783 prefix: str = "",
1784 is_last: bool = True,
1785 first: bool = True,
1786 indent_size: int = 4,
1787 ) -> str:
1788 """
1789 Return a tree-like structure of the filesystem starting from the given path as a string.
1791 Parameters
1792 ----------
1793 path: Root path to start traversal from
1794 recursion_limit: Maximum depth of directory traversal
1795 max_display: Maximum number of items to display per directory
1796 display_size: Whether to display file sizes
1797 prefix: Current line prefix for visual tree structure
1798 is_last: Whether current item is last in its level
1799 first: Whether this is the first call (displays root path)
1800 indent_size: Number of spaces by indent
1802 Returns
1803 -------
1804 str: A string representing the tree structure.
1806 Example
1807 -------
1808 >>> from fsspec import filesystem
1810 >>> fs = filesystem('ftp', host='test.rebex.net', user='demo', password='password')
1811 >>> tree = fs.tree(display_size=True, recursion_limit=3, indent_size=8, max_display=10)
1812 >>> print(tree)
1813 """
1815 def format_bytes(n: int) -> str:
1816 """Format bytes as text."""
1817 for prefix, k in (
1818 ("P", 2**50),
1819 ("T", 2**40),
1820 ("G", 2**30),
1821 ("M", 2**20),
1822 ("k", 2**10),
1823 ):
1824 if n >= 0.9 * k:
1825 return f"{n / k:.2f} {prefix}b"
1826 return f"{n}B"
1828 result = []
1830 if first:
1831 result.append(path)
1833 if recursion_limit:
1834 indent = " " * indent_size
1835 contents = self.ls(path, detail=True)
1836 contents.sort(
1837 key=lambda x: (x.get("type") != "directory", x.get("name", ""))
1838 )
1840 if max_display is not None and len(contents) > max_display:
1841 displayed_contents = contents[:max_display]
1842 remaining_count = len(contents) - max_display
1843 else:
1844 displayed_contents = contents
1845 remaining_count = 0
1847 for i, item in enumerate(displayed_contents):
1848 is_last_item = (i == len(displayed_contents) - 1) and (
1849 remaining_count == 0
1850 )
1852 branch = (
1853 "└" + ("─" * (indent_size - 2))
1854 if is_last_item
1855 else "├" + ("─" * (indent_size - 2))
1856 )
1857 branch += " "
1858 new_prefix = prefix + (
1859 indent if is_last_item else "│" + " " * (indent_size - 1)
1860 )
1862 name = os.path.basename(item.get("name", ""))
1864 if display_size and item.get("type") == "directory":
1865 sub_contents = self.ls(item.get("name", ""), detail=True)
1866 num_files = sum(
1867 1 for sub_item in sub_contents if sub_item.get("type") == "file"
1868 )
1869 num_folders = sum(
1870 1
1871 for sub_item in sub_contents
1872 if sub_item.get("type") == "directory"
1873 )
1875 if num_files == 0 and num_folders == 0:
1876 size = " (empty folder)"
1877 elif num_files == 0:
1878 size = f" ({num_folders} subfolder{'s' if num_folders > 1 else ''})"
1879 elif num_folders == 0:
1880 size = f" ({num_files} file{'s' if num_files > 1 else ''})"
1881 else:
1882 size = f" ({num_files} file{'s' if num_files > 1 else ''}, {num_folders} subfolder{'s' if num_folders > 1 else ''})"
1883 elif display_size and item.get("type") == "file":
1884 size = f" ({format_bytes(item.get('size', 0))})"
1885 else:
1886 size = ""
1888 result.append(f"{prefix}{branch}{name}{size}")
1890 if item.get("type") == "directory" and recursion_limit > 0:
1891 result.append(
1892 self.tree(
1893 path=item.get("name", ""),
1894 recursion_limit=recursion_limit - 1,
1895 max_display=max_display,
1896 display_size=display_size,
1897 prefix=new_prefix,
1898 is_last=is_last_item,
1899 first=False,
1900 indent_size=indent_size,
1901 )
1902 )
1904 if remaining_count > 0:
1905 more_message = f"{remaining_count} more item(s) not displayed."
1906 result.append(
1907 f"{prefix}{'└' + ('─' * (indent_size - 2))} {more_message}"
1908 )
1910 return "\n".join(_ for _ in result if _)
1912 # ------------------------------------------------------------------------
1913 # Aliases
1915 def read_bytes(self, path, start=None, end=None, **kwargs):
1916 """Alias of `AbstractFileSystem.cat_file`."""
1917 return self.cat_file(path, start=start, end=end, **kwargs)
1919 def write_bytes(self, path, value, **kwargs):
1920 """Alias of `AbstractFileSystem.pipe_file`."""
1921 self.pipe_file(path, value, **kwargs)
1923 def makedir(self, path, create_parents=True, **kwargs):
1924 """Alias of `AbstractFileSystem.mkdir`."""
1925 return self.mkdir(path, create_parents=create_parents, **kwargs)
1927 def mkdirs(self, path, exist_ok=False):
1928 """Alias of `AbstractFileSystem.makedirs`."""
1929 return self.makedirs(path, exist_ok=exist_ok)
1931 def listdir(self, path, detail=True, **kwargs):
1932 """Alias of `AbstractFileSystem.ls`."""
1933 return self.ls(path, detail=detail, **kwargs)
1935 def cp(self, path1, path2, **kwargs):
1936 """Alias of `AbstractFileSystem.copy`."""
1937 return self.copy(path1, path2, **kwargs)
1939 def move(self, path1, path2, **kwargs):
1940 """Alias of `AbstractFileSystem.mv`."""
1941 return self.mv(path1, path2, **kwargs)
1943 def stat(self, path, **kwargs):
1944 """Alias of `AbstractFileSystem.info`."""
1945 return self.info(path, **kwargs)
1947 def disk_usage(self, path, total=True, maxdepth=None, **kwargs):
1948 """Alias of `AbstractFileSystem.du`."""
1949 return self.du(path, total=total, maxdepth=maxdepth, **kwargs)
1951 def rename(self, path1, path2, **kwargs):
1952 """Alias of `AbstractFileSystem.mv`."""
1953 return self.mv(path1, path2, **kwargs)
1955 def delete(self, path, recursive=False, maxdepth=None):
1956 """Alias of `AbstractFileSystem.rm`."""
1957 return self.rm(path, recursive=recursive, maxdepth=maxdepth)
1959 def upload(self, lpath, rpath, recursive=False, **kwargs):
1960 """Alias of `AbstractFileSystem.put`."""
1961 return self.put(lpath, rpath, recursive=recursive, **kwargs)
1963 def download(self, rpath, lpath, recursive=False, **kwargs):
1964 """Alias of `AbstractFileSystem.get`."""
1965 return self.get(rpath, lpath, recursive=recursive, **kwargs)
1967 def sign(self, path, expiration=100, **kwargs):
1968 """Create a signed URL representing the given path
1970 Some implementations allow temporary URLs to be generated, as a
1971 way of delegating credentials.
1973 Parameters
1974 ----------
1975 path : str
1976 The path on the filesystem
1977 expiration : int
1978 Number of seconds to enable the URL for (if supported)
1980 Returns
1981 -------
1982 URL : str
1983 The signed URL
1985 Raises
1986 ------
1987 NotImplementedError : if method is not implemented for a filesystem
1988 """
1989 raise NotImplementedError("Sign is not implemented for this filesystem")
1991 def _isfilestore(self):
1992 # Originally inherited from pyarrow DaskFileSystem. Keeping this
1993 # here for backwards compatibility as long as pyarrow uses its
1994 # legacy fsspec-compatible filesystems and thus accepts fsspec
1995 # filesystems as well
1996 return False
1999class AbstractBufferedFile(io.IOBase):
2000 """Convenient class to derive from to provide buffering
2002 In the case that the backend does not provide a pythonic file-like object
2003 already, this class contains much of the logic to build one. The only
2004 methods that need to be overridden are ``_upload_chunk``,
2005 ``_initiate_upload`` and ``_fetch_range``.
2006 """
2008 DEFAULT_BLOCK_SIZE = 5 * 2**20
2009 _details = None
2011 def __init__(
2012 self,
2013 fs,
2014 path,
2015 mode="rb",
2016 block_size="default",
2017 autocommit=True,
2018 cache_type="readahead",
2019 cache_options=None,
2020 size=None,
2021 **kwargs,
2022 ):
2023 """
2024 Template for files with buffered reading and writing
2026 Parameters
2027 ----------
2028 fs: instance of FileSystem
2029 path: str
2030 location in file-system
2031 mode: str
2032 Normal file modes. Currently only 'wb', 'ab' or 'rb'. Some file
2033 systems may be read-only, and some may not support append.
2034 block_size: int
2035 Buffer size for reading or writing, 'default' for class default
2036 autocommit: bool
2037 Whether to write to final destination; may only impact what
2038 happens when file is being closed.
2039 cache_type: {"readahead", "none", "mmap", "bytes"}, default "readahead"
2040 Caching policy in read mode. See the definitions in ``core``.
2041 cache_options : dict
2042 Additional options passed to the constructor for the cache specified
2043 by `cache_type`.
2044 size: int
2045 If given and in read mode, suppressed having to look up the file size
2046 kwargs:
2047 Gets stored as self.kwargs
2048 """
2049 from .core import caches
2051 self.path = path
2052 self.fs = fs
2053 self.mode = mode
2054 self.blocksize = (
2055 self.DEFAULT_BLOCK_SIZE if block_size in ["default", None] else block_size
2056 )
2057 self.loc = 0
2058 self.autocommit = autocommit
2059 self.end = None
2060 self.start = None
2061 self.closed = False
2063 if cache_options is None:
2064 cache_options = {}
2066 if "trim" in kwargs:
2067 warnings.warn(
2068 "Passing 'trim' to control the cache behavior has been deprecated. "
2069 "Specify it within the 'cache_options' argument instead.",
2070 FutureWarning,
2071 )
2072 cache_options["trim"] = kwargs.pop("trim")
2074 self.kwargs = kwargs
2076 if mode not in {"ab", "rb", "wb", "xb"}:
2077 raise NotImplementedError("File mode not supported")
2078 if mode == "rb":
2079 if size is not None:
2080 self.size = size
2081 else:
2082 self.size = self.details["size"]
2083 self.cache = caches[cache_type](
2084 self.blocksize, self._fetch_range, self.size, **cache_options
2085 )
2086 else:
2087 self.buffer = io.BytesIO()
2088 self.offset = None
2089 self.forced = False
2090 self.location = None
2092 @property
2093 def len(self):
2094 """Total size in bytes, independent of the current file position.
2096 Raises AttributeError when the backend has not provided a known size.
2097 """
2098 if self.size is None:
2099 raise AttributeError("File size is unknown")
2100 return self.size
2102 @property
2103 def details(self):
2104 if self._details is None:
2105 self._details = self.fs.info(self.path)
2106 return self._details
2108 @details.setter
2109 def details(self, value):
2110 self._details = value
2111 self.size = value["size"]
2113 @property
2114 def full_name(self):
2115 return _unstrip_protocol(self.path, self.fs)
2117 @property
2118 def closed(self):
2119 # get around this attr being read-only in IOBase
2120 # use getattr here, since this can be called during del
2121 return getattr(self, "_closed", True)
2123 @closed.setter
2124 def closed(self, c):
2125 self._closed = c
2127 def __hash__(self):
2128 if "w" in self.mode:
2129 return id(self)
2130 else:
2131 return int(tokenize(self.details), 16)
2133 def __eq__(self, other):
2134 """Files are equal if they have the same checksum, only in read mode"""
2135 if self is other:
2136 return True
2137 return (
2138 isinstance(other, type(self))
2139 and self.mode == "rb"
2140 and other.mode == "rb"
2141 and hash(self) == hash(other)
2142 )
2144 def commit(self):
2145 """Move from temp to final destination"""
2147 def discard(self):
2148 """Throw away temporary file"""
2150 def info(self):
2151 """File information about this path"""
2152 if self.readable():
2153 return self.details
2154 else:
2155 raise ValueError("Info not available while writing")
2157 def tell(self):
2158 """Current file location"""
2159 return self.loc
2161 def seek(self, loc, whence=0):
2162 """Set current file location
2164 Parameters
2165 ----------
2166 loc: int
2167 byte location
2168 whence: {0, 1, 2}
2169 from start of file, current location or end of file, resp.
2170 """
2171 loc = int(loc)
2172 if not self.mode == "rb":
2173 raise OSError(ESPIPE, "Seek only available in read mode")
2174 if whence == 0:
2175 nloc = loc
2176 elif whence == 1:
2177 nloc = self.loc + loc
2178 elif whence == 2:
2179 nloc = self.size + loc
2180 else:
2181 raise ValueError(f"invalid whence ({whence}, should be 0, 1 or 2)")
2182 if nloc < 0:
2183 raise ValueError("Seek before start of file")
2184 self.loc = nloc
2185 return self.loc
2187 def write(self, data):
2188 """
2189 Write data to buffer.
2191 Buffer only sent on flush() or if buffer is greater than
2192 or equal to blocksize.
2194 Parameters
2195 ----------
2196 data: bytes
2197 Set of bytes to be written.
2198 """
2199 if not self.writable():
2200 raise ValueError("File not in write mode")
2201 if self.closed:
2202 raise ValueError("I/O operation on closed file.")
2203 if self.forced:
2204 raise ValueError("This file has been force-flushed, can only close")
2205 out = self.buffer.write(data)
2206 self.loc += out
2207 if self.buffer.tell() >= self.blocksize:
2208 self.flush()
2209 return out
2211 def flush(self, force=False):
2212 """
2213 Write buffered data to backend store.
2215 Writes the current buffer, if it is larger than the block-size, or if
2216 the file is being closed.
2218 Parameters
2219 ----------
2220 force: bool
2221 When closing, write the last block even if it is smaller than
2222 blocks are allowed to be. Disallows further writing to this file.
2223 """
2225 if self.closed:
2226 raise ValueError("Flush on closed file")
2227 if force and self.forced:
2228 raise ValueError("Force flush cannot be called more than once")
2229 if force:
2230 self.forced = True
2232 if self.readable():
2233 # no-op to flush on read-mode
2234 return
2236 if not force and self.buffer.tell() < self.blocksize:
2237 # Defer write on small block
2238 return
2240 if self.offset is None:
2241 # Initialize a multipart upload
2242 self.offset = 0
2243 try:
2244 self._initiate_upload()
2245 except Exception:
2246 self.closed = True
2247 raise
2249 if self._upload_chunk(final=force) is not False:
2250 self.offset += self.buffer.seek(0, 2)
2251 self.buffer = io.BytesIO()
2253 def _upload_chunk(self, final=False):
2254 """Write one part of a multi-block file upload
2256 Parameters
2257 ==========
2258 final: bool
2259 This is the last block, so should complete file, if
2260 self.autocommit is True.
2261 """
2262 # may not yet have been initialized, may need to call _initialize_upload
2264 def _initiate_upload(self):
2265 """Create remote file/upload"""
2266 pass
2268 def _fetch_range(self, start, end):
2269 """Get the specified set of bytes from remote"""
2270 return self.fs.cat_file(self.path, start=start, end=end)
2272 def read(self, length=-1):
2273 """
2274 Return data from cache, or fetch pieces as necessary
2276 Parameters
2277 ----------
2278 length: int (-1)
2279 Number of bytes to read; if <0, all remaining bytes.
2280 """
2281 length = -1 if length is None else int(length)
2282 if self.mode != "rb":
2283 raise ValueError("File not in read mode")
2284 if length < 0:
2285 length = self.size - self.loc
2286 if self.closed:
2287 raise ValueError("I/O operation on closed file.")
2288 if length == 0:
2289 # don't even bother calling fetch
2290 return b""
2291 out = self.cache._fetch(self.loc, self.loc + length)
2293 logger.debug(
2294 "%s read: %i - %i %s",
2295 self,
2296 self.loc,
2297 self.loc + length,
2298 self.cache._log_stats(),
2299 )
2300 self.loc += len(out)
2301 return out
2303 def readinto(self, b):
2304 """mirrors builtin file's readinto method
2306 https://docs.python.org/3/library/io.html#io.RawIOBase.readinto
2307 """
2308 out = memoryview(b).cast("B")
2309 if out.readonly:
2310 self.read(0)
2311 raise TypeError("readinto() argument must be read-write bytes-like object")
2312 data = self.read(out.nbytes)
2313 out[: len(data)] = data
2314 return len(data)
2316 def readuntil(self, char=b"\n", blocks=None):
2317 """Return data between current position and first occurrence of char
2319 char is included in the output, except if the end of the tile is
2320 encountered first.
2322 Parameters
2323 ----------
2324 char: bytes
2325 Thing to find
2326 blocks: None or int
2327 How much to read in each go. Defaults to file blocksize - which may
2328 mean a new read on every call.
2329 """
2330 out = []
2331 tail = b""
2332 overlap = max(len(char) - 1, 0)
2333 while True:
2334 start = self.tell()
2335 part = self.read(blocks or self.blocksize)
2336 if len(part) == 0:
2337 break
2338 data = tail + part
2339 found = data.find(char)
2340 if found > -1:
2341 consumed = found + len(char) - len(tail)
2342 out.append(part[:consumed])
2343 self.seek(start + consumed)
2344 break
2345 out.append(part)
2346 tail = data[-overlap:] if overlap else b""
2347 return b"".join(out)
2349 def readline(self):
2350 """Read until and including the first occurrence of newline character
2352 Note that, because of character encoding, this is not necessarily a
2353 true line ending.
2354 """
2355 return self.readuntil(b"\n")
2357 def __next__(self):
2358 out = self.readline()
2359 if out:
2360 return out
2361 raise StopIteration
2363 def __iter__(self):
2364 return self
2366 def readlines(self):
2367 """Return all data, split by the newline character, including the newline character"""
2368 data = self.read()
2369 if not data:
2370 return []
2371 lines = data.split(b"\n")
2372 out = [l + b"\n" for l in lines[:-1]]
2373 if data.endswith(b"\n"):
2374 return out
2375 else:
2376 return out + [lines[-1]]
2377 # return list(self) ???
2379 def readinto1(self, b):
2380 return self.readinto(b)
2382 def close(self):
2383 """Close file
2385 Finalizes writes, discards cache
2386 """
2387 if getattr(self, "_unclosable", False):
2388 return
2389 if self.closed:
2390 return
2391 try:
2392 if self.mode == "rb":
2393 cache = getattr(self, "cache", None)
2394 if cache is not None:
2395 close = getattr(cache, "close", None)
2396 if callable(close):
2397 close()
2398 self.cache = None
2399 else:
2400 if not getattr(self, "forced", True):
2401 self.flush(force=True)
2403 if self.fs is not None:
2404 self.fs.invalidate_cache(self.path)
2405 self.fs.invalidate_cache(self.fs._parent(self.path))
2406 finally:
2407 self.closed = True
2409 def readable(self):
2410 """Whether opened for reading"""
2411 return "r" in self.mode and not self.closed
2413 def seekable(self):
2414 """Whether is seekable (only in read mode)"""
2415 return self.readable()
2417 def writable(self):
2418 """Whether opened for writing"""
2419 return self.mode in {"wb", "ab", "xb"} and not self.closed
2421 def __reduce__(self):
2422 if self.mode != "rb":
2423 raise RuntimeError("Pickling a writeable file is not supported")
2425 return reopen, (
2426 self.fs,
2427 self.path,
2428 self.mode,
2429 self.blocksize,
2430 self.loc,
2431 self.size,
2432 self.autocommit,
2433 self.cache.name if self.cache else "none",
2434 self.kwargs,
2435 )
2437 def __del__(self):
2438 if not self.closed:
2439 self.close()
2441 def __str__(self):
2442 return f"<File-like object {type(self.fs).__name__}, {self.path}>"
2444 __repr__ = __str__
2446 def __enter__(self):
2447 return self
2449 def __exit__(self, *args):
2450 self.close()
2453def reopen(fs, path, mode, blocksize, loc, size, autocommit, cache_type, kwargs):
2454 file = fs.open(
2455 path,
2456 mode=mode,
2457 block_size=blocksize,
2458 autocommit=autocommit,
2459 cache_type=cache_type,
2460 size=size,
2461 **kwargs,
2462 )
2463 if loc > 0:
2464 file.seek(loc)
2465 return file