from__future__importannotationsimportioimportjsonimportloggingimportosimportthreadingimportwarningsimportweakreffromerrnoimportESPIPEfromglobimporthas_magicfromhashlibimportsha256fromtypingimportAny,ClassVarfrom.callbacksimportDEFAULT_CALLBACKfrom.configimportapply_config,conffrom.dircacheimportDirCachefrom.transactionimportTransactionfrom.utilsimport(_unstrip_protocol,glob_translate,isfilelike,other_paths,read_block,stringify_path,tokenize,)logger=logging.getLogger("fsspec")defmake_instance(cls,args,kwargs):returncls(*args,**kwargs)class_Cached(type):"""    Metaclass for caching file system instances.    Notes    -----    Instances are cached according to    * The values of the class attributes listed in `_extra_tokenize_attributes`    * The arguments passed to ``__init__``.    This creates an additional reference to the filesystem, which prevents the    filesystem from being garbage collected when all *user* references go away.    A call to the :meth:`AbstractFileSystem.clear_instance_cache` must *also*    be made for a filesystem instance to be garbage collected.    """def__init__(cls,*args,**kwargs):super().__init__(*args,**kwargs)# Note: we intentionally create a reference here, to avoid garbage# collecting instances when all other references are gone. To really# delete a FileSystem, the cache must be cleared.ifconf.get("weakref_instance_cache"):# pragma: no cover# debug option for analysing fork/spawn conditionscls._cache=weakref.WeakValueDictionary()else:cls._cache={}cls._pid=os.getpid()def__call__(cls,*args,**kwargs):kwargs=apply_config(cls,kwargs)extra_tokens=tuple(getattr(cls,attr,None)forattrincls._extra_tokenize_attributes)strip_tokenize_options={k:kwargs.pop(k)forkincls._strip_tokenize_optionsifkinkwargs}token=tokenize(cls,cls._pid,threading.get_ident(),*args,*extra_tokens,**kwargs)skip=kwargs.pop("skip_instance_cache",False)ifos.getpid()!=cls._pid:cls._cache.clear()cls._pid=os.getpid()ifnotskipandcls.cachableandtokenincls._cache:cls._latest=tokenreturncls._cache[token]else:obj=super().__call__(*args,**kwargs,**strip_tokenize_options)# Setting _fs_token here causes some static linters to complain.obj._fs_token_=tokenobj.storage_args=argsobj.storage_options=kwargsifobj.async_implandobj.mirror_sync_methods:from.asynimportmirror_sync_methodsmirror_sync_methods(obj)ifcls.cachableandnotskip:cls._latest=tokencls._cache[token]=objreturnobj
[docs]classAbstractFileSystem(metaclass=_Cached):"""    An abstract super-class for pythonic file-systems    Implementations are expected to be compatible with or, better, subclass    from here.    """cachable=True# this class can be cached, instances reused_cached=Falseblocksize=2**22sep="/"protocol:ClassVar[str|tuple[str,...]]="abstract"_latest=Noneasync_impl=Falsemirror_sync_methods=Falseroot_marker=""# For some FSs, may require leading '/' or other charactertransaction_type=Transaction#: Extra *class attributes* that should be considered when hashing._extra_tokenize_attributes=()#: *storage options* that should not be considered when hashing._strip_tokenize_options=()# Set by _Cached metaclassstorage_args:tuple[Any,...]storage_options:dict[str,Any]def__init__(self,*args,**storage_options):"""Create and configure file-system instance        Instances may be cachable, so if similar enough arguments are seen        a new instance is not required. The token attribute exists to allow        implementations to cache instances if they wish.        A reasonable default should be provided if there are no arguments.        Subclasses should call this method.        Parameters        ----------        use_listings_cache, listings_expiry_time, max_paths:            passed to ``DirCache``, if the implementation supports            directory listing caching. Pass use_listings_cache=False            to disable such caching.        skip_instance_cache: bool            If this is a cachable implementation, pass True here to force            creating a new instance even if a matching instance exists, and prevent            storing this instance.        asynchronous: bool        loop: asyncio-compatible IOLoop or None        """ifself._cached:# reusing instance, don't changereturnself._cached=Trueself._intrans=Falseself._transaction=Noneself._invalidated_caches_in_transaction=[]self.dircache=DirCache(**storage_options)ifstorage_options.pop("add_docs",None):warnings.warn("add_docs is no longer supported.",FutureWarning)ifstorage_options.pop("add_aliases",None):warnings.warn("add_aliases has been removed.",FutureWarning)# This is set in _Cachedself._fs_token_=None@propertydeffsid(self):"""Persistent filesystem id that can be used to compare filesystems        across sessions.        """raiseNotImplementedError@propertydef_fs_token(self):returnself._fs_token_def__dask_tokenize__(self):returnself._fs_tokendef__hash__(self):returnint(self._fs_token,16)def__eq__(self,other):returnisinstance(other,type(self))andself._fs_token==other._fs_tokendef__reduce__(self):returnmake_instance,(type(self),self.storage_args,self.storage_options)@classmethoddef_strip_protocol(cls,path):"""Turn path from fully-qualified to file-system-specific        May require FS-specific handling, e.g., for relative paths or links.        """ifisinstance(path,list):return[cls._strip_protocol(p)forpinpath]path=stringify_path(path)protos=(cls.protocol,)ifisinstance(cls.protocol,str)elsecls.protocolforprotocolinprotos:ifpath.startswith(protocol+"://"):path=path[len(protocol)+3:]elifpath.startswith(protocol+"::"):path=path[len(protocol)+2:]path=path.rstrip("/")# use of root_marker to make minimum required path, e.g., "/"returnpathorcls.root_marker
[docs]defunstrip_protocol(self,name:str)->str:"""Format FS-specific path to generic, including protocol"""protos=(self.protocol,)ifisinstance(self.protocol,str)elseself.protocolforprotocolinprotos:ifname.startswith(f"{protocol}://"):returnnamereturnf"{protos[0]}://{name}"
@staticmethoddef_get_kwargs_from_urls(path):"""If kwargs can be encoded in the paths, extract them here        This should happen before instantiation of the class; incoming paths        then should be amended to strip the options in methods.        Examples may look like an sftp path "sftp://user@host:/my/path", where        the user and host should become kwargs and later get stripped.        """# by default, nothing happensreturn{}
[docs]@classmethoddefcurrent(cls):"""Return the most recently instantiated FileSystem        If no instance has been created, then create one with defaults        """ifcls._latestincls._cache:returncls._cache[cls._latest]returncls()
@propertydeftransaction(self):"""A context within which files are committed together upon exit        Requires the file class to implement `.commit()` and `.discard()`        for the normal and exception cases.        """ifself._transactionisNone:self._transaction=self.transaction_type(self)returnself._transaction
[docs]defstart_transaction(self):"""Begin write transaction for deferring files, non-context version"""self._intrans=Trueself._transaction=self.transaction_type(self)returnself.transaction
[docs]defend_transaction(self):"""Finish write transaction, non-context version"""self.transaction.complete()self._transaction=None# The invalid cache must be cleared after the transaction is completed.forpathinself._invalidated_caches_in_transaction:self.invalidate_cache(path)self._invalidated_caches_in_transaction.clear()
[docs]definvalidate_cache(self,path=None):"""        Discard any cached directory information        Parameters        ----------        path: string or None            If None, clear all listings cached else listings at or under given            path.        """# Not necessary to implement invalidation mechanism, may have no cache.# But if have, you should call this method of parent class from your# subclass to ensure expiring caches after transacations correctly.# See the implementation of FTPFileSystem in ftp.pyifself._intrans:self._invalidated_caches_in_transaction.append(path)
[docs]defmkdir(self,path,create_parents=True,**kwargs):"""        Create directory entry at path        For systems that don't have true directories, may create an for        this instance only and not touch the real filesystem        Parameters        ----------        path: str            location        create_parents: bool            if True, this is equivalent to ``makedirs``        kwargs:            may be permissions, etc.        """pass# not necessary to implement, may not have directories
[docs]defmakedirs(self,path,exist_ok=False):"""Recursively make directories        Creates directory at path and any intervening required directories.        Raises exception if, for instance, the path already exists but is a        file.        Parameters        ----------        path: str            leaf directory name        exist_ok: bool (False)            If False, will error if the target already exists        """pass# not necessary to implement, may not have directories
[docs]defrmdir(self,path):"""Remove a directory, if empty"""pass# not necessary to implement, may not have directories
[docs]defls(self,path,detail=True,**kwargs):"""List objects at path.        This should include subdirectories and files at that location. The        difference between a file and a directory must be clear when details        are requested.        The specific keys, or perhaps a FileInfo class, or similar, is TBD,        but must be consistent across implementations.        Must include:        - full path to the entry (without protocol)        - size of the entry, in bytes. If the value cannot be determined, will          be ``None``.        - type of entry, "file", "directory" or other        Additional information        may be present, appropriate to the file-system, e.g., generation,        checksum, etc.        May use refresh=True|False to allow use of self._ls_from_cache to        check for a saved listing and avoid calling the backend. This would be        common where listing may be expensive.        Parameters        ----------        path: str        detail: bool            if True, gives a list of dictionaries, where each is the same as            the result of ``info(path)``. If False, gives a list of paths            (str).        kwargs: may have additional backend-specific options, such as version            information        Returns        -------        List of strings if detail is False, or list of directory information        dicts if detail is True.        """raiseNotImplementedError
def_ls_from_cache(self,path):"""Check cache for listing        Returns listing, if found (may be empty list for a directly that exists        but contains nothing), None if not in cache.        """parent=self._parent(path)try:returnself.dircache[path.rstrip("/")]exceptKeyError:passtry:files=[fforfinself.dircache[parent]iff["name"]==pathor(f["name"]==path.rstrip("/")andf["type"]=="directory")]iflen(files)==0:# parent dir was listed but did not contain this fileraiseFileNotFoundError(path)returnfilesexceptKeyError:pass
[docs]defwalk(self,path,maxdepth=None,topdown=True,on_error="omit",**kwargs):"""Return all files under the given path.        List all files, recursing into subdirectories; output is iterator-style,        like ``os.walk()``. For a simple list of files, ``find()`` is available.        When topdown is True, the caller can modify the dirnames list in-place (perhaps        using del or slice assignment), and walk() will        only recurse into the subdirectories whose names remain in dirnames;        this can be used to prune the search, impose a specific order of visiting,        or even to inform walk() about directories the caller creates or renames before        it resumes walk() again.        Modifying dirnames when topdown is False has no effect. (see os.walk)        Note that the "files" outputted will include anything that is not        a directory, such as links.        Parameters        ----------        path: str            Root to recurse into        maxdepth: int            Maximum recursion depth. None means limitless, but not recommended            on link-based file-systems.        topdown: bool (True)            Whether to walk the directory tree from the top downwards or from            the bottom upwards.        on_error: "omit", "raise", a callable            if omit (default), path with exception will simply be empty;            If raise, an underlying exception will be raised;            if callable, it will be called with a single OSError instance as argument        kwargs: passed to ``ls``        """ifmaxdepthisnotNoneandmaxdepth<1:raiseValueError("maxdepth must be at least 1")path=self._strip_protocol(path)full_dirs={}dirs={}files={}detail=kwargs.pop("detail",False)try:listing=self.ls(path,detail=True,**kwargs)except(FileNotFoundError,OSError)ase:ifon_error=="raise":raiseifcallable(on_error):on_error(e)returnforinfoinlisting:# each info name must be at least [path]/part , but here# we check also for names like [path]/part/pathname=info["name"].rstrip("/")name=pathname.rsplit("/",1)[-1]ifinfo["type"]=="directory"andpathname!=path:# do not include "self" pathfull_dirs[name]=pathnamedirs[name]=infoelifpathname==path:# file-like with same name as give pathfiles[""]=infoelse:files[name]=infoifnotdetail:dirs=list(dirs)files=list(files)iftopdown:# Yield before recursion if walking top downyieldpath,dirs,filesifmaxdepthisnotNone:maxdepth-=1ifmaxdepth<1:ifnottopdown:yieldpath,dirs,filesreturnfordindirs:yield fromself.walk(full_dirs[d],maxdepth=maxdepth,detail=detail,topdown=topdown,**kwargs,)ifnottopdown:# Yield after recursion if walking bottom upyieldpath,dirs,files
[docs]deffind(self,path,maxdepth=None,withdirs=False,detail=False,**kwargs):"""List all files below path.        Like posix ``find`` command without conditions        Parameters        ----------        path : str        maxdepth: int or None            If not None, the maximum number of levels to descend        withdirs: bool            Whether to include directory paths in the output. This is True            when used by glob, but users usually only want files.        kwargs are passed to ``ls``.        """# TODO: allow equivalent of -name parameterpath=self._strip_protocol(path)out={}# Add the root directory if withdirs is requested# This is needed for posix glob complianceifwithdirsandpath!=""andself.isdir(path):out[path]=self.info(path)for_,dirs,filesinself.walk(path,maxdepth,detail=True,**kwargs):ifwithdirs:files.update(dirs)out.update({info["name"]:infoforname,infoinfiles.items()})ifnotoutandself.isfile(path):# walk works on directories, but find should also return [path]# when path happens to be a fileout[path]={}names=sorted(out)ifnotdetail:returnnameselse:return{name:out[name]fornameinnames}
[docs]defdu(self,path,total=True,maxdepth=None,withdirs=False,**kwargs):"""Space used by files and optionally directories within a path        Directory size does not include the size of its contents.        Parameters        ----------        path: str        total: bool            Whether to sum all the file sizes        maxdepth: int or None            Maximum number of directory levels to descend, None for unlimited.        withdirs: bool            Whether to include directory paths in the output.        kwargs: passed to ``find``        Returns        -------        Dict of {path: size} if total=False, or int otherwise, where numbers        refer to bytes used.        """sizes={}ifwithdirsandself.isdir(path):# Include top-level directory in outputinfo=self.info(path)sizes[info["name"]]=info["size"]forfinself.find(path,maxdepth=maxdepth,withdirs=withdirs,**kwargs):info=self.info(f)sizes[info["name"]]=info["size"]iftotal:returnsum(sizes.values())else:returnsizes
[docs]defglob(self,path,maxdepth=None,**kwargs):"""Find files by glob-matching.        Pattern matching capabilities for finding files that match the given pattern.        Parameters        ----------        path: str            The glob pattern to match against        maxdepth: int or None            Maximum depth for ``'**'`` patterns. Applied on the first ``'**'`` found.            Must be at least 1 if provided.        kwargs:            Additional arguments passed to ``find`` (e.g., detail=True)        Returns        -------        List of matched paths, or dict of paths and their info if detail=True        Notes        -----        Supported patterns:        - '*': Matches any sequence of characters within a single directory level        - ``'**'``: Matches any number of directory levels (must be an entire path component)        - '?': Matches exactly one character        - '[abc]': Matches any character in the set        - '[a-z]': Matches any character in the range        - '[!abc]': Matches any character NOT in the set        Special behaviors:        - If the path ends with '/', only folders are returned        - Consecutive '*' characters are compressed into a single '*'        - Empty brackets '[]' never match anything        - Negated empty brackets '[!]' match any single character        - Special characters in character classes are escaped properly        Limitations:        - ``'**'`` must be a complete path component (e.g., ``'a/**/b'``, not ``'a**b'``)        - No brace expansion ('{a,b}.txt')        - No extended glob patterns ('+(pattern)', '!(pattern)')        """ifmaxdepthisnotNoneandmaxdepth<1:raiseValueError("maxdepth must be at least 1")importreseps=(os.path.sep,os.path.altsep)ifos.path.altsepelse(os.path.sep,)ends_with_sep=path.endswith(seps)# _strip_protocol strips trailing slashpath=self._strip_protocol(path)append_slash_to_dirname=ends_with_seporpath.endswith(tuple(sep+"**"forsepinseps))idx_star=path.find("*")ifpath.find("*")>=0elselen(path)idx_qmark=path.find("?")ifpath.find("?")>=0elselen(path)idx_brace=path.find("[")ifpath.find("[")>=0elselen(path)min_idx=min(idx_star,idx_qmark,idx_brace)detail=kwargs.pop("detail",False)ifnothas_magic(path):ifself.exists(path,**kwargs):ifnotdetail:return[path]else:return{path:self.info(path,**kwargs)}else:ifnotdetail:return[]# glob of non-existent returns emptyelse:return{}elif"/"inpath[:min_idx]:min_idx=path[:min_idx].rindex("/")root=path[:min_idx+1]depth=path[min_idx+1:].count("/")+1else:root=""depth=path[min_idx+1:].count("/")+1if"**"inpath:ifmaxdepthisnotNone:idx_double_stars=path.find("**")depth_double_stars=path[idx_double_stars:].count("/")+1depth=depth-depth_double_stars+maxdepthelse:depth=Noneallpaths=self.find(root,maxdepth=depth,withdirs=True,detail=True,**kwargs)pattern=glob_translate(path+("/"ifends_with_sepelse""))pattern=re.compile(pattern)out={p:infoforp,infoinsorted(allpaths.items())ifpattern.match(p+"/"ifappend_slash_to_dirnameandinfo["type"]=="directory"elsep)}ifdetail:returnoutelse:returnlist(out)
[docs]defexists(self,path,**kwargs):"""Is there a file at the given path"""try:self.info(path,**kwargs)returnTrueexcept:# noqa: E722# any exception allowed bar FileNotFoundError?returnFalse
[docs]deflexists(self,path,**kwargs):"""If there is a file at the given path (including        broken links)"""returnself.exists(path)
[docs]definfo(self,path,**kwargs):"""Give details of entry at path        Returns a single dictionary, with exactly the same information as ``ls``        would with ``detail=True``.        The default implementation calls ls and could be overridden by a        shortcut. kwargs are passed on to ```ls()``.        Some file systems might not be able to measure the file's size, in        which case, the returned dict will include ``'size': None``.        Returns        -------        dict with keys: name (full path in the FS), size (in bytes), type (file,        directory, or something else) and other FS-specific keys.        """path=self._strip_protocol(path)out=self.ls(self._parent(path),detail=True,**kwargs)out=[oforoinoutifo["name"].rstrip("/")==path]ifout:returnout[0]out=self.ls(path,detail=True,**kwargs)path=path.rstrip("/")out1=[oforoinoutifo["name"].rstrip("/")==path]iflen(out1)==1:if"size"notinout1[0]:out1[0]["size"]=Nonereturnout1[0]eliflen(out1)>1orout:return{"name":path,"size":0,"type":"directory"}else:raiseFileNotFoundError(path)
[docs]defchecksum(self,path):"""Unique value for current version of file        If the checksum is the same from one moment to another, the contents        are guaranteed to be the same. If the checksum changes, the contents        *might* have changed.        This should normally be overridden; default will probably capture        creation/modification timestamp (which would be good) or maybe        access timestamp (which would be bad)        """returnint(tokenize(self.info(path)),16)
[docs]defsize(self,path):"""Size in bytes of file"""returnself.info(path).get("size",None)
[docs]defsizes(self,paths):"""Size in bytes of each file in a list of paths"""return[self.size(p)forpinpaths]
[docs]defisdir(self,path):"""Is this entry directory-like?"""try:returnself.info(path)["type"]=="directory"exceptOSError:returnFalse
[docs]defisfile(self,path):"""Is this entry file-like?"""try:returnself.info(path)["type"]=="file"except:# noqa: E722returnFalse
[docs]defread_text(self,path,encoding=None,errors=None,newline=None,**kwargs):"""Get the contents of the file as a string.        Parameters        ----------        path: str            URL of file on this filesystems        encoding, errors, newline: same as `open`.        """withself.open(path,mode="r",encoding=encoding,errors=errors,newline=newline,**kwargs,)asf:returnf.read()
[docs]defwrite_text(self,path,value,encoding=None,errors=None,newline=None,**kwargs):"""Write the text to the given file.        An existing file will be overwritten.        Parameters        ----------        path: str            URL of file on this filesystems        value: str            Text to write.        encoding, errors, newline: same as `open`.        """withself.open(path,mode="w",encoding=encoding,errors=errors,newline=newline,**kwargs,)asf:returnf.write(value)
[docs]defcat_file(self,path,start=None,end=None,**kwargs):"""Get the content of a file        Parameters        ----------        path: URL of file on this filesystems        start, end: int            Bytes limits of the read. If negative, backwards from end,            like usual python slices. Either can be None for start or            end of file, respectively        kwargs: passed to ``open()``.        """# explicitly set buffering off?withself.open(path,"rb",**kwargs)asf:ifstartisnotNone:ifstart>=0:f.seek(start)else:f.seek(max(0,f.size+start))ifendisnotNone:ifend<0:end=f.size+endreturnf.read(end-f.tell())returnf.read()
[docs]defpipe_file(self,path,value,mode="overwrite",**kwargs):"""Set the bytes of given file"""ifmode=="create"andself.exists(path):# non-atomic but simple way; or could use "xb" in open(), which is likely# not as well supportedraiseFileExistsErrorwithself.open(path,"wb",**kwargs)asf:f.write(value)
[docs]defpipe(self,path,value=None,**kwargs):"""Put value into path        (counterpart to ``cat``)        Parameters        ----------        path: string or dict(str, bytes)            If a string, a single remote location to put ``value`` bytes; if a dict,            a mapping of {path: bytesvalue}.        value: bytes, optional            If using a single path, these are the bytes to put there. Ignored if            ``path`` is a dict        """ifisinstance(path,str):self.pipe_file(self._strip_protocol(path),value,**kwargs)elifisinstance(path,dict):fork,vinpath.items():self.pipe_file(self._strip_protocol(k),v,**kwargs)else:raiseValueError("path must be str or dict")
[docs]defcat_ranges(self,paths,starts,ends,max_gap=None,on_error="return",**kwargs):"""Get the contents of byte ranges from one or more files        Parameters        ----------        paths: list            A list of of filepaths on this filesystems        starts, ends: int or list            Bytes limits of the read. If using a single int, the same value will be            used to read all the specified files.        """ifmax_gapisnotNone:raiseNotImplementedErrorifnotisinstance(paths,list):raiseTypeErrorifnotisinstance(starts,list):starts=[starts]*len(paths)ifnotisinstance(ends,list):ends=[ends]*len(paths)iflen(starts)!=len(paths)orlen(ends)!=len(paths):raiseValueErrorout=[]forp,s,einzip(paths,starts,ends):try:out.append(self.cat_file(p,s,e))exceptExceptionase:ifon_error=="return":out.append(e)else:raisereturnout
[docs]defcat(self,path,recursive=False,on_error="raise",**kwargs):"""Fetch (potentially multiple) paths' contents        Parameters        ----------        recursive: bool            If True, assume the path(s) are directories, and get all the            contained files        on_error : "raise", "omit", "return"            If raise, an underlying exception will be raised (converted to KeyError            if the type is in self.missing_exceptions); if omit, keys with exception            will simply not be included in the output; if "return", all keys are            included in the output, but the value will be bytes or an exception            instance.        kwargs: passed to cat_file        Returns        -------        dict of {path: contents} if there are multiple paths        or the path has been otherwise expanded        """paths=self.expand_path(path,recursive=recursive,**kwargs)if(len(paths)>1orisinstance(path,list)orpaths[0]!=self._strip_protocol(path)):out={}forpathinpaths:try:out[path]=self.cat_file(path,**kwargs)exceptExceptionase:ifon_error=="raise":raiseifon_error=="return":out[path]=ereturnoutelse:returnself.cat_file(paths[0],**kwargs)
[docs]defget_file(self,rpath,lpath,callback=DEFAULT_CALLBACK,outfile=None,**kwargs):"""Copy single remote file to local"""from.implementations.localimportLocalFileSystemifisfilelike(lpath):outfile=lpathelifself.isdir(rpath):os.makedirs(lpath,exist_ok=True)returnNonefs=LocalFileSystem(auto_mkdir=True)fs.makedirs(fs._parent(lpath),exist_ok=True)withself.open(rpath,"rb",**kwargs)asf1:ifoutfileisNone:outfile=open(lpath,"wb")try:callback.set_size(getattr(f1,"size",None))data=Truewhiledata:data=f1.read(self.blocksize)segment_len=outfile.write(data)ifsegment_lenisNone:segment_len=len(data)callback.relative_update(segment_len)finally:ifnotisfilelike(lpath):outfile.close()
[docs]defget(self,rpath,lpath,recursive=False,callback=DEFAULT_CALLBACK,maxdepth=None,**kwargs,):"""Copy file(s) to local.        Copies a specific file or tree of files (if recursive=True). If lpath        ends with a "/", it will be assumed to be a directory, and target files        will go within. Can submit a list of paths, which may be glob-patterns        and will be expanded.        Calls get_file for each source.        """ifisinstance(lpath,list)andisinstance(rpath,list):# No need to expand paths when both source and destination# are provided as listsrpaths=rpathlpaths=lpathelse:from.implementations.localimport(LocalFileSystem,make_path_posix,trailing_sep,)source_is_str=isinstance(rpath,str)rpaths=self.expand_path(rpath,recursive=recursive,maxdepth=maxdepth,**kwargs)ifsource_is_strand(notrecursiveormaxdepthisnotNone):# Non-recursive glob does not copy directoriesrpaths=[pforpinrpathsifnot(trailing_sep(p)orself.isdir(p))]ifnotrpaths:returnifisinstance(lpath,str):lpath=make_path_posix(lpath)source_is_file=len(rpaths)==1dest_is_dir=isinstance(lpath,str)and(trailing_sep(lpath)orLocalFileSystem().isdir(lpath))exists=source_is_strand((has_magic(rpath)andsource_is_file)or(nothas_magic(rpath)anddest_is_dirandnottrailing_sep(rpath)))lpaths=other_paths(rpaths,lpath,exists=exists,flatten=notsource_is_str,)callback.set_size(len(lpaths))forlpath,rpathincallback.wrap(zip(lpaths,rpaths)):withcallback.branched(rpath,lpath)aschild:self.get_file(rpath,lpath,callback=child,**kwargs)
[docs]defput_file(self,lpath,rpath,callback=DEFAULT_CALLBACK,mode="overwrite",**kwargs):"""Copy single file to remote"""ifmode=="create"andself.exists(rpath):raiseFileExistsErrorifos.path.isdir(lpath):self.makedirs(rpath,exist_ok=True)returnNonewithopen(lpath,"rb")asf1:size=f1.seek(0,2)callback.set_size(size)f1.seek(0)self.mkdirs(self._parent(os.fspath(rpath)),exist_ok=True)withself.open(rpath,"wb",**kwargs)asf2:whilef1.tell()<size:data=f1.read(self.blocksize)segment_len=f2.write(data)ifsegment_lenisNone:segment_len=len(data)callback.relative_update(segment_len)
[docs]defput(self,lpath,rpath,recursive=False,callback=DEFAULT_CALLBACK,maxdepth=None,**kwargs,):"""Copy file(s) from local.        Copies a specific file or tree of files (if recursive=True). If rpath        ends with a "/", it will be assumed to be a directory, and target files        will go within.        Calls put_file for each source.        """ifisinstance(lpath,list)andisinstance(rpath,list):# No need to expand paths when both source and destination# are provided as listsrpaths=rpathlpaths=lpathelse:from.implementations.localimport(LocalFileSystem,make_path_posix,trailing_sep,)source_is_str=isinstance(lpath,str)ifsource_is_str:lpath=make_path_posix(lpath)fs=LocalFileSystem()lpaths=fs.expand_path(lpath,recursive=recursive,maxdepth=maxdepth,**kwargs)ifsource_is_strand(notrecursiveormaxdepthisnotNone):# Non-recursive glob does not copy directorieslpaths=[pforpinlpathsifnot(trailing_sep(p)orfs.isdir(p))]ifnotlpaths:returnsource_is_file=len(lpaths)==1dest_is_dir=isinstance(rpath,str)and(trailing_sep(rpath)orself.isdir(rpath))rpath=(self._strip_protocol(rpath)ifisinstance(rpath,str)else[self._strip_protocol(p)forpinrpath])exists=source_is_strand((has_magic(lpath)andsource_is_file)or(nothas_magic(lpath)anddest_is_dirandnottrailing_sep(lpath)))rpaths=other_paths(lpaths,rpath,exists=exists,flatten=notsource_is_str,)callback.set_size(len(rpaths))forlpath,rpathincallback.wrap(zip(lpaths,rpaths)):withcallback.branched(lpath,rpath)aschild:self.put_file(lpath,rpath,callback=child,**kwargs)
[docs]defhead(self,path,size=1024):"""Get the first ``size`` bytes from file"""withself.open(path,"rb")asf:returnf.read(size)
[docs]deftail(self,path,size=1024):"""Get the last ``size`` bytes from file"""withself.open(path,"rb")asf:f.seek(max(-size,-f.size),2)returnf.read()
defcp_file(self,path1,path2,**kwargs):raiseNotImplementedError
[docs]defcopy(self,path1,path2,recursive=False,maxdepth=None,on_error=None,**kwargs):"""Copy within two locations in the filesystem        on_error : "raise", "ignore"            If raise, any not-found exceptions will be raised; if ignore any            not-found exceptions will cause the path to be skipped; defaults to            raise unless recursive is true, where the default is ignore        """ifon_errorisNoneandrecursive:on_error="ignore"elifon_errorisNone:on_error="raise"ifisinstance(path1,list)andisinstance(path2,list):# No need to expand paths when both source and destination# are provided as listspaths1=path1paths2=path2else:from.implementations.localimporttrailing_sepsource_is_str=isinstance(path1,str)paths1=self.expand_path(path1,recursive=recursive,maxdepth=maxdepth,**kwargs)ifsource_is_strand(notrecursiveormaxdepthisnotNone):# Non-recursive glob does not copy directoriespaths1=[pforpinpaths1ifnot(trailing_sep(p)orself.isdir(p))]ifnotpaths1:returnsource_is_file=len(paths1)==1dest_is_dir=isinstance(path2,str)and(trailing_sep(path2)orself.isdir(path2))exists=source_is_strand((has_magic(path1)andsource_is_file)or(nothas_magic(path1)anddest_is_dirandnottrailing_sep(path1)))paths2=other_paths(paths1,path2,exists=exists,flatten=notsource_is_str,)forp1,p2inzip(paths1,paths2):try:self.cp_file(p1,p2,**kwargs)exceptFileNotFoundError:ifon_error=="raise":raise
[docs]defexpand_path(self,path,recursive=False,maxdepth=None,**kwargs):"""Turn one or more globs or directories into a list of all matching paths        to files or directories.        kwargs are passed to ``glob`` or ``find``, which may in turn call ``ls``        """ifmaxdepthisnotNoneandmaxdepth<1:raiseValueError("maxdepth must be at least 1")ifisinstance(path,(str,os.PathLike)):out=self.expand_path([path],recursive,maxdepth,**kwargs)else:out=set()path=[self._strip_protocol(p)forpinpath]forpinpath:ifhas_magic(p):bit=set(self.glob(p,maxdepth=maxdepth,**kwargs))out|=bitifrecursive:# glob call above expanded one depth so if maxdepth is defined# then decrement it in expand_path call below. If it is zero# after decrementing then avoid expand_path call.ifmaxdepthisnotNoneandmaxdepth<=1:continueout|=set(self.expand_path(list(bit),recursive=recursive,maxdepth=maxdepth-1ifmaxdepthisnotNoneelseNone,**kwargs,))continueelifrecursive:rec=set(self.find(p,maxdepth=maxdepth,withdirs=True,detail=False,**kwargs))out|=recifpnotinoutand(recursiveisFalseorself.exists(p)):# should only check once, for the rootout.add(p)ifnotout:raiseFileNotFoundError(path)returnsorted(out)
[docs]defmv(self,path1,path2,recursive=False,maxdepth=None,**kwargs):"""Move file(s) from one location to another"""ifpath1==path2:logger.debug("%s mv: The paths are the same, so no files were moved.",self)else:# explicitly raise exception to prevent data corruptionself.copy(path1,path2,recursive=recursive,maxdepth=maxdepth,onerror="raise")self.rm(path1,recursive=recursive)
[docs]defrm_file(self,path):"""Delete a file"""self._rm(path)
def_rm(self,path):"""Delete one file"""# this is the old name for the method, prefer rm_fileraiseNotImplementedError
[docs]defrm(self,path,recursive=False,maxdepth=None):"""Delete files.        Parameters        ----------        path: str or list of str            File(s) to delete.        recursive: bool            If file(s) are directories, recursively delete contents and then            also remove the directory        maxdepth: int or None            Depth to pass to walk for finding files to delete, if recursive.            If None, there will be no limit and infinite recursion may be            possible.        """path=self.expand_path(path,recursive=recursive,maxdepth=maxdepth)forpinreversed(path):self.rm_file(p)
@classmethoddef_parent(cls,path):path=cls._strip_protocol(path)if"/"inpath:parent=path.rsplit("/",1)[0].lstrip(cls.root_marker)returncls.root_marker+parentelse:returncls.root_markerdef_open(self,path,mode="rb",block_size=None,autocommit=True,cache_options=None,**kwargs,):"""Return raw bytes-mode file-like from the file-system"""returnAbstractBufferedFile(self,path,mode,block_size,autocommit,cache_options=cache_options,**kwargs,)
[docs]defopen(self,path,mode="rb",block_size=None,cache_options=None,compression=None,**kwargs,):"""        Return a file-like object from the filesystem        The resultant instance must function correctly in a context ``with``        block.        Parameters        ----------        path: str            Target file        mode: str like 'rb', 'w'            See builtin ``open()``            Mode "x" (exclusive write) may be implemented by the backend. Even if            it is, whether  it is checked up front or on commit, and whether it is            atomic is implementation-dependent.        block_size: int            Some indication of buffering - this is a value in bytes        cache_options : dict, optional            Extra arguments to pass through to the cache.        compression: string or None            If given, open file using compression codec. Can either be a compression            name (a key in ``fsspec.compression.compr``) or "infer" to guess the            compression from the filename suffix.        encoding, errors, newline: passed on to TextIOWrapper for text mode        """importiopath=self._strip_protocol(path)if"b"notinmode:mode=mode.replace("t","")+"b"text_kwargs={k:kwargs.pop(k)forkin["encoding","errors","newline"]ifkinkwargs}returnio.TextIOWrapper(self.open(path,mode,block_size=block_size,cache_options=cache_options,compression=compression,**kwargs,),**text_kwargs,)else:ac=kwargs.pop("autocommit",notself._intrans)f=self._open(path,mode=mode,block_size=block_size,autocommit=ac,cache_options=cache_options,**kwargs,)ifcompressionisnotNone:fromfsspec.compressionimportcomprfromfsspec.coreimportget_compressioncompression=get_compression(path,compression)compress=compr[compression]f=compress(f,mode=mode[0])ifnotacand"r"notinmode:self.transaction.files.append(f)returnf
[docs]deftouch(self,path,truncate=True,**kwargs):"""Create empty file, or update timestamp        Parameters        ----------        path: str            file location        truncate: bool            If True, always set file size to 0; if False, update timestamp and            leave file unchanged, if backend allows this        """iftruncateornotself.exists(path):withself.open(path,"wb",**kwargs):passelse:raiseNotImplementedError# update timestamp, if possible
[docs]defukey(self,path):"""Hash of file properties, to tell if it has changed"""returnsha256(str(self.info(path)).encode()).hexdigest()
[docs]defread_block(self,fn,offset,length,delimiter=None):"""Read a block of bytes from        Starting at ``offset`` of the file, read ``length`` bytes.  If        ``delimiter`` is set then we ensure that the read starts and stops at        delimiter boundaries that follow the locations ``offset`` and ``offset        + length``.  If ``offset`` is zero then we start at zero.  The        bytestring returned WILL include the end delimiter string.        If offset+length is beyond the eof, reads to eof.        Parameters        ----------        fn: string            Path to filename        offset: int            Byte offset to start read        length: int            Number of bytes to read. If None, read to end.        delimiter: bytes (optional)            Ensure reading starts and stops at delimiter bytestring        Examples        --------        >>> fs.read_block('data/file.csv', 0, 13)  # doctest: +SKIP        b'Alice, 100\\nBo'        >>> fs.read_block('data/file.csv', 0, 13, delimiter=b'\\n')  # doctest: +SKIP        b'Alice, 100\\nBob, 200\\n'        Use ``length=None`` to read to the end of the file.        >>> fs.read_block('data/file.csv', 0, None, delimiter=b'\\n')  # doctest: +SKIP        b'Alice, 100\\nBob, 200\\nCharlie, 300'        See Also        --------        :func:`fsspec.utils.read_block`        """withself.open(fn,"rb")asf:size=f.sizeiflengthisNone:length=sizeifsizeisnotNoneandoffset+length>size:length=size-offsetreturnread_block(f,offset,length,delimiter)
[docs]defto_json(self,*,include_password:bool=True)->str:"""        JSON representation of this filesystem instance.        Parameters        ----------        include_password: bool, default True            Whether to include the password (if any) in the output.        Returns        -------        JSON string with keys ``cls`` (the python location of this class),        protocol (text name of this class's protocol, first one in case of        multiple), ``args`` (positional args, usually empty), and all other        keyword arguments as their own keys.        Warnings        --------        Serialized filesystems may contain sensitive information which have been        passed to the constructor, such as passwords and tokens. Make sure you        store and send them in a secure environment!        """from.jsonimportFilesystemJSONEncoderreturnjson.dumps(self,cls=type("_FilesystemJSONEncoder",(FilesystemJSONEncoder,),{"include_password":include_password},),)
[docs]@staticmethoddeffrom_json(blob:str)->AbstractFileSystem:"""        Recreate a filesystem instance from JSON representation.        See ``.to_json()`` for the expected structure of the input.        Parameters        ----------        blob: str        Returns        -------        file system instance, not necessarily of this particular class.        Warnings        --------        This can import arbitrary modules (as determined by the ``cls`` key).        Make sure you haven't installed any modules that may execute malicious code        at import time.        """from.jsonimportFilesystemJSONDecoderreturnjson.loads(blob,cls=FilesystemJSONDecoder)
[docs]defto_dict(self,*,include_password:bool=True)->dict[str,Any]:"""        JSON-serializable dictionary representation of this filesystem instance.        Parameters        ----------        include_password: bool, default True            Whether to include the password (if any) in the output.        Returns        -------        Dictionary with keys ``cls`` (the python location of this class),        protocol (text name of this class's protocol, first one in case of        multiple), ``args`` (positional args, usually empty), and all other        keyword arguments as their own keys.        Warnings        --------        Serialized filesystems may contain sensitive information which have been        passed to the constructor, such as passwords and tokens. Make sure you        store and send them in a secure environment!        """from.jsonimportFilesystemJSONEncoderjson_encoder=FilesystemJSONEncoder()cls=type(self)proto=self.protocolstorage_options=dict(self.storage_options)ifnotinclude_password:storage_options.pop("password",None)returndict(cls=f"{cls.__module__}:{cls.__name__}",protocol=proto[0]ifisinstance(proto,(tuple,list))elseproto,args=json_encoder.make_serializable(self.storage_args),**json_encoder.make_serializable(storage_options),)
[docs]@staticmethoddeffrom_dict(dct:dict[str,Any])->AbstractFileSystem:"""        Recreate a filesystem instance from dictionary representation.        See ``.to_dict()`` for the expected structure of the input.        Parameters        ----------        dct: Dict[str, Any]        Returns        -------        file system instance, not necessarily of this particular class.        Warnings        --------        This can import arbitrary modules (as determined by the ``cls`` key).        Make sure you haven't installed any modules that may execute malicious code        at import time.        """from.jsonimportFilesystemJSONDecoderjson_decoder=FilesystemJSONDecoder()dct=dict(dct)# Defensive copycls=FilesystemJSONDecoder.try_resolve_fs_cls(dct)ifclsisNone:raiseValueError("Not a serialized AbstractFileSystem")dct.pop("cls",None)dct.pop("protocol",None)returncls(*json_decoder.unmake_serializable(dct.pop("args",())),**json_decoder.unmake_serializable(dct),)
def_get_pyarrow_filesystem(self):"""        Make a version of the FS instance which will be acceptable to pyarrow        """# all instances already also derive from pyarrowreturnself
[docs]defget_mapper(self,root="",check=False,create=False,missing_exceptions=None):"""Create key/value store based on this file-system        Makes a MutableMapping interface to the FS at the given root path.        See ``fsspec.mapping.FSMap`` for further details.        """from.mappingimportFSMapreturnFSMap(root,self,check=check,create=create,missing_exceptions=missing_exceptions,)
[docs]@classmethoddefclear_instance_cache(cls):"""        Clear the cache of filesystem instances.        Notes        -----        Unless overridden by setting the ``cachable`` class attribute to False,        the filesystem class stores a reference to newly created instances. This        prevents Python's normal rules around garbage collection from working,        since the instances refcount will not drop to zero until        ``clear_instance_cache`` is called.        """cls._cache.clear()
[docs]defcreated(self,path):"""Return the created timestamp of a file as a datetime.datetime"""raiseNotImplementedError
[docs]defmodified(self,path):"""Return the modified timestamp of a file as a datetime.datetime"""raiseNotImplementedError
[docs]deftree(self,path:str="/",recursion_limit:int=2,max_display:int=25,display_size:bool=False,prefix:str="",is_last:bool=True,first:bool=True,indent_size:int=4,)->str:"""        Return a tree-like structure of the filesystem starting from the given path as a string.        Parameters        ----------            path: Root path to start traversal from            recursion_limit: Maximum depth of directory traversal            max_display: Maximum number of items to display per directory            display_size: Whether to display file sizes            prefix: Current line prefix for visual tree structure            is_last: Whether current item is last in its level            first: Whether this is the first call (displays root path)            indent_size: Number of spaces by indent        Returns        -------            str: A string representing the tree structure.        Example        -------            >>> from fsspec import filesystem            >>> fs = filesystem('ftp', host='test.rebex.net', user='demo', password='password')            >>> tree = fs.tree(display_size=True, recursion_limit=3, indent_size=8, max_display=10)            >>> print(tree)        """defformat_bytes(n:int)->str:"""Format bytes as text."""forprefix,kin(("P",2**50),("T",2**40),("G",2**30),("M",2**20),("k",2**10),):ifn>=0.9*k:returnf"{n/k:.2f}{prefix}b"returnf"{n}B"result=[]iffirst:result.append(path)ifrecursion_limit:indent=" "*indent_sizecontents=self.ls(path,detail=True)contents.sort(key=lambdax:(x.get("type")!="directory",x.get("name","")))ifmax_displayisnotNoneandlen(contents)>max_display:displayed_contents=contents[:max_display]remaining_count=len(contents)-max_displayelse:displayed_contents=contentsremaining_count=0fori,iteminenumerate(displayed_contents):is_last_item=(i==len(displayed_contents)-1)and(remaining_count==0)branch=("└"+("─"*(indent_size-2))ifis_last_itemelse"├"+("─"*(indent_size-2)))branch+=" "new_prefix=prefix+(indentifis_last_itemelse"│"+" "*(indent_size-1))name=os.path.basename(item.get("name",""))ifdisplay_sizeanditem.get("type")=="directory":sub_contents=self.ls(item.get("name",""),detail=True)num_files=sum(1forsub_iteminsub_contentsifsub_item.get("type")=="file")num_folders=sum(1forsub_iteminsub_contentsifsub_item.get("type")=="directory")ifnum_files==0andnum_folders==0:size=" (empty folder)"elifnum_files==0:size=f" ({num_folders} subfolder{'s'ifnum_folders>1else''})"elifnum_folders==0:size=f" ({num_files} file{'s'ifnum_files>1else''})"else:size=f" ({num_files} file{'s'ifnum_files>1else''},{num_folders} subfolder{'s'ifnum_folders>1else''})"elifdisplay_sizeanditem.get("type")=="file":size=f" ({format_bytes(item.get('size',0))})"else:size=""result.append(f"{prefix}{branch}{name}{size}")ifitem.get("type")=="directory"andrecursion_limit>0:result.append(self.tree(path=item.get("name",""),recursion_limit=recursion_limit-1,max_display=max_display,display_size=display_size,prefix=new_prefix,is_last=is_last_item,first=False,indent_size=indent_size,))ifremaining_count>0:more_message=f"{remaining_count} more item(s) not displayed."result.append(f"{prefix}{'└'+('─'*(indent_size-2))}{more_message}")return"\n".join(_for_inresultif_)
# ------------------------------------------------------------------------# Aliases
[docs]defread_bytes(self,path,start=None,end=None,**kwargs):"""Alias of `AbstractFileSystem.cat_file`."""returnself.cat_file(path,start=start,end=end,**kwargs)
[docs]defwrite_bytes(self,path,value,**kwargs):"""Alias of `AbstractFileSystem.pipe_file`."""self.pipe_file(path,value,**kwargs)
[docs]defmakedir(self,path,create_parents=True,**kwargs):"""Alias of `AbstractFileSystem.mkdir`."""returnself.mkdir(path,create_parents=create_parents,**kwargs)
[docs]defmkdirs(self,path,exist_ok=False):"""Alias of `AbstractFileSystem.makedirs`."""returnself.makedirs(path,exist_ok=exist_ok)
[docs]deflistdir(self,path,detail=True,**kwargs):"""Alias of `AbstractFileSystem.ls`."""returnself.ls(path,detail=detail,**kwargs)
[docs]defcp(self,path1,path2,**kwargs):"""Alias of `AbstractFileSystem.copy`."""returnself.copy(path1,path2,**kwargs)
[docs]defmove(self,path1,path2,**kwargs):"""Alias of `AbstractFileSystem.mv`."""returnself.mv(path1,path2,**kwargs)
[docs]defstat(self,path,**kwargs):"""Alias of `AbstractFileSystem.info`."""returnself.info(path,**kwargs)
[docs]defdisk_usage(self,path,total=True,maxdepth=None,**kwargs):"""Alias of `AbstractFileSystem.du`."""returnself.du(path,total=total,maxdepth=maxdepth,**kwargs)
[docs]defrename(self,path1,path2,**kwargs):"""Alias of `AbstractFileSystem.mv`."""returnself.mv(path1,path2,**kwargs)
[docs]defdelete(self,path,recursive=False,maxdepth=None):"""Alias of `AbstractFileSystem.rm`."""returnself.rm(path,recursive=recursive,maxdepth=maxdepth)
[docs]defupload(self,lpath,rpath,recursive=False,**kwargs):"""Alias of `AbstractFileSystem.put`."""returnself.put(lpath,rpath,recursive=recursive,**kwargs)
[docs]defdownload(self,rpath,lpath,recursive=False,**kwargs):"""Alias of `AbstractFileSystem.get`."""returnself.get(rpath,lpath,recursive=recursive,**kwargs)
[docs]defsign(self,path,expiration=100,**kwargs):"""Create a signed URL representing the given path        Some implementations allow temporary URLs to be generated, as a        way of delegating credentials.        Parameters        ----------        path : str             The path on the filesystem        expiration : int            Number of seconds to enable the URL for (if supported)        Returns        -------        URL : str            The signed URL        Raises        ------        NotImplementedError : if method is not implemented for a filesystem        """raiseNotImplementedError("Sign is not implemented for this filesystem")
def_isfilestore(self):# Originally inherited from pyarrow DaskFileSystem. Keeping this# here for backwards compatibility as long as pyarrow uses its# legacy fsspec-compatible filesystems and thus accepts fsspec# filesystems as wellreturnFalse
[docs]classAbstractBufferedFile(io.IOBase):"""Convenient class to derive from to provide buffering    In the case that the backend does not provide a pythonic file-like object    already, this class contains much of the logic to build one. The only    methods that need to be overridden are ``_upload_chunk``,    ``_initiate_upload`` and ``_fetch_range``.    """DEFAULT_BLOCK_SIZE=5*2**20_details=Nonedef__init__(self,fs,path,mode="rb",block_size="default",autocommit=True,cache_type="readahead",cache_options=None,size=None,**kwargs,):"""        Template for files with buffered reading and writing        Parameters        ----------        fs: instance of FileSystem        path: str            location in file-system        mode: str            Normal file modes. Currently only 'wb', 'ab' or 'rb'. Some file            systems may be read-only, and some may not support append.        block_size: int            Buffer size for reading or writing, 'default' for class default        autocommit: bool            Whether to write to final destination; may only impact what            happens when file is being closed.        cache_type: {"readahead", "none", "mmap", "bytes"}, default "readahead"            Caching policy in read mode. See the definitions in ``core``.        cache_options : dict            Additional options passed to the constructor for the cache specified            by `cache_type`.        size: int            If given and in read mode, suppressed having to look up the file size        kwargs:            Gets stored as self.kwargs        """from.coreimportcachesself.path=pathself.fs=fsself.mode=modeself.blocksize=(self.DEFAULT_BLOCK_SIZEifblock_sizein["default",None]elseblock_size)self.loc=0self.autocommit=autocommitself.end=Noneself.start=Noneself.closed=Falseifcache_optionsisNone:cache_options={}if"trim"inkwargs:warnings.warn("Passing 'trim' to control the cache behavior has been deprecated. ""Specify it within the 'cache_options' argument instead.",FutureWarning,)cache_options["trim"]=kwargs.pop("trim")self.kwargs=kwargsifmodenotin{"ab","rb","wb","xb"}:raiseNotImplementedError("File mode not supported")ifmode=="rb":ifsizeisnotNone:self.size=sizeelse:self.size=self.details["size"]self.cache=caches[cache_type](self.blocksize,self._fetch_range,self.size,**cache_options)else:self.buffer=io.BytesIO()self.offset=Noneself.forced=Falseself.location=None@propertydefdetails(self):ifself._detailsisNone:self._details=self.fs.info(self.path)returnself._details@details.setterdefdetails(self,value):self._details=valueself.size=value["size"]@propertydeffull_name(self):return_unstrip_protocol(self.path,self.fs)@propertydefclosed(self):# get around this attr being read-only in IOBase# use getattr here, since this can be called during delreturngetattr(self,"_closed",True)@closed.setterdefclosed(self,c):self._closed=cdef__hash__(self):if"w"inself.mode:returnid(self)else:returnint(tokenize(self.details),16)def__eq__(self,other):"""Files are equal if they have the same checksum, only in read mode"""ifselfisother:returnTruereturn(isinstance(other,type(self))andself.mode=="rb"andother.mode=="rb"andhash(self)==hash(other))
[docs]defcommit(self):"""Move from temp to final destination"""
[docs]defdiscard(self):"""Throw away temporary file"""
[docs]definfo(self):"""File information about this path"""ifself.readable():returnself.detailselse:raiseValueError("Info not available while writing")
[docs]deftell(self):"""Current file location"""returnself.loc
[docs]defseek(self,loc,whence=0):"""Set current file location        Parameters        ----------        loc: int            byte location        whence: {0, 1, 2}            from start of file, current location or end of file, resp.        """loc=int(loc)ifnotself.mode=="rb":raiseOSError(ESPIPE,"Seek only available in read mode")ifwhence==0:nloc=locelifwhence==1:nloc=self.loc+locelifwhence==2:nloc=self.size+locelse:raiseValueError(f"invalid whence ({whence}, should be 0, 1 or 2)")ifnloc<0:raiseValueError("Seek before start of file")self.loc=nlocreturnself.loc
[docs]defwrite(self,data):"""        Write data to buffer.        Buffer only sent on flush() or if buffer is greater than        or equal to blocksize.        Parameters        ----------        data: bytes            Set of bytes to be written.        """ifnotself.writable():raiseValueError("File not in write mode")ifself.closed:raiseValueError("I/O operation on closed file.")ifself.forced:raiseValueError("This file has been force-flushed, can only close")out=self.buffer.write(data)self.loc+=outifself.buffer.tell()>=self.blocksize:self.flush()returnout
[docs]defflush(self,force=False):"""        Write buffered data to backend store.        Writes the current buffer, if it is larger than the block-size, or if        the file is being closed.        Parameters        ----------        force: bool            When closing, write the last block even if it is smaller than            blocks are allowed to be. Disallows further writing to this file.        """ifself.closed:raiseValueError("Flush on closed file")ifforceandself.forced:raiseValueError("Force flush cannot be called more than once")ifforce:self.forced=Trueifself.readable():# no-op to flush on read-modereturnifnotforceandself.buffer.tell()<self.blocksize:# Defer write on small blockreturnifself.offsetisNone:# Initialize a multipart uploadself.offset=0try:self._initiate_upload()except:self.closed=Trueraiseifself._upload_chunk(final=force)isnotFalse:self.offset+=self.buffer.seek(0,2)self.buffer=io.BytesIO()
def_upload_chunk(self,final=False):"""Write one part of a multi-block file upload        Parameters        ==========        final: bool            This is the last block, so should complete file, if            self.autocommit is True.        """# may not yet have been initialized, may need to call _initialize_uploaddef_initiate_upload(self):"""Create remote file/upload"""passdef_fetch_range(self,start,end):"""Get the specified set of bytes from remote"""returnself.fs.cat_file(self.path,start=start,end=end)
[docs]defread(self,length=-1):"""        Return data from cache, or fetch pieces as necessary        Parameters        ----------        length: int (-1)            Number of bytes to read; if <0, all remaining bytes.        """length=-1iflengthisNoneelseint(length)ifself.mode!="rb":raiseValueError("File not in read mode")iflength<0:length=self.size-self.locifself.closed:raiseValueError("I/O operation on closed file.")iflength==0:# don't even bother calling fetchreturnb""out=self.cache._fetch(self.loc,self.loc+length)logger.debug("%s read:%i -%i%s",self,self.loc,self.loc+length,self.cache._log_stats(),)self.loc+=len(out)returnout
[docs]defreadinto(self,b):"""mirrors builtin file's readinto method        https://docs.python.org/3/library/io.html#io.RawIOBase.readinto        """out=memoryview(b).cast("B")data=self.read(out.nbytes)out[:len(data)]=datareturnlen(data)
[docs]defreaduntil(self,char=b"\n",blocks=None):"""Return data between current position and first occurrence of char        char is included in the output, except if the end of the tile is        encountered first.        Parameters        ----------        char: bytes            Thing to find        blocks: None or int            How much to read in each go. Defaults to file blocksize - which may            mean a new read on every call.        """out=[]whileTrue:start=self.tell()part=self.read(blocksorself.blocksize)iflen(part)==0:breakfound=part.find(char)iffound>-1:out.append(part[:found+len(char)])self.seek(start+found+len(char))breakout.append(part)returnb"".join(out)
[docs]defreadline(self):"""Read until and including the first occurrence of newline character        Note that, because of character encoding, this is not necessarily a        true line ending.        """returnself.readuntil(b"\n")
def__next__(self):out=self.readline()ifout:returnoutraiseStopIterationdef__iter__(self):returnself
[docs]defreadlines(self):"""Return all data, split by the newline character, including the newline character"""data=self.read()lines=data.split(b"\n")out=[l+b"\n"forlinlines[:-1]]ifdata.endswith(b"\n"):returnoutelse:returnout+[lines[-1]]
# return list(self)  ???defreadinto1(self,b):returnself.readinto(b)
[docs]defclose(self):"""Close file        Finalizes writes, discards cache        """ifgetattr(self,"_unclosable",False):returnifself.closed:returntry:ifself.mode=="rb":self.cache=Noneelse:ifnotself.forced:self.flush(force=True)ifself.fsisnotNone:self.fs.invalidate_cache(self.path)self.fs.invalidate_cache(self.fs._parent(self.path))finally:self.closed=True
[docs]defreadable(self):"""Whether opened for reading"""return"r"inself.modeandnotself.closed
[docs]defseekable(self):"""Whether is seekable (only in read mode)"""returnself.readable()
[docs]defwritable(self):"""Whether opened for writing"""returnself.modein{"wb","ab","xb"}andnotself.closed
def__reduce__(self):ifself.mode!="rb":raiseRuntimeError("Pickling a writeable file is not supported")returnreopen,(self.fs,self.path,self.mode,self.blocksize,self.loc,self.size,self.autocommit,self.cache.nameifself.cacheelse"none",self.kwargs,)def__del__(self):ifnotself.closed:self.close()def__str__(self):returnf"<File-like object{type(self.fs).__name__},{self.path}>"__repr__=__str__def__enter__(self):returnselfdef__exit__(self,*args):self.close()
defreopen(fs,path,mode,blocksize,loc,size,autocommit,cache_type,kwargs):file=fs.open(path,mode=mode,block_size=blocksize,autocommit=autocommit,cache_type=cache_type,size=size,**kwargs,)ifloc>0:file.seek(loc)returnfile