Repository: fsspec/filesystem_spec
Pull request #1716 (2024-10-12 18:05:14)

Apply ruff/isort rules (I)


Patch:
diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 9797e6a..f9e5ed9 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -13,7 +13,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        PY: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        PY:
+          - "3.9"
+          - "3.10"
+          - "3.11"
+          - "3.12"
+          - "3.13"
 
     env:
       CIRUN: true
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 48e3c95..f1f11e9 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -5,7 +5,7 @@ exclude: >
 repos:
 
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v5.0.0
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer
@@ -14,7 +14,7 @@ repos:
       - id: check-yaml
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.4.4
+    rev: v0.6.9
     hooks:
       # Run the linter.
       - id: ruff
diff --git a/docs/source/api.rst b/docs/source/api.rst
index 9fdcd1f..5385acc 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -209,41 +209,73 @@ Built-in Implementations
 Other Known Implementations
 ---------------------------
 
-- `abfs`_ for Azure Blob service
-- `adl`_ for Azure DataLake storage
+
+Note that most of these projects are hosted outside of the `fsspec` organisation. Please read their
+documentation carefully before using any particular package.
+
+- `abfs`_ for Azure Blob service, with protocol "abfs://"
+- `adl`_ for Azure DataLake storage, with protocol "adl://"
 - `alluxiofs`_ to access fsspec implemented filesystem with Alluxio distributed cache
-- `boxfs`_ for access to Box file storage
-- `dropbox`_ for access to dropbox shares
+- `boxfs`_ for access to Box file storage, with protocol "box://"
+- `csvbase`_ for access to csvbase.com hosted CSV files, with protocol "csvbase://"
+- `dropbox`_ for access to dropbox shares, with protocol "dropbox://"
 - `dvc`_ to access DVC/Git repository as a filesystem
-- `gcsfs`_ for Google Cloud Storage
+- `fsspec-encrypted`_ for transparent encryption on top of other fsspec filesystems.
+- `gcsfs`_ for Google Cloud Storage, with protocol "gcs://"
 - `gdrive`_ to access Google Drive and shares (experimental)
+- `git`_ to access Git repositories
 - `huggingface_hub`_ to access the Hugging Face Hub filesystem, with protocol "hf://"
-- `lakefs`_ for lakeFS data lakes
-- `ocifs`_ for access to Oracle Cloud Object Storage
+- `hdfs-native`_ to access Hadoop filesystem, with protocol "hdfs://"
+- `httpfs-sync`_ to access HTTP(s) files in a synchronous manner to offer an alternative to the aiohttp-based implementation.
+- `ipfsspec`_ for the InterPlanetary File System (IPFS), with protocol "ipfs://"
+- `irods`_ for access to iRODS servers, with protocol "irods://"
+- `lakefs`_ for lakeFS data lakes, with protocol "lakefs://"
+- `morefs`_ for `OverlayFileSystem`, `DictFileSystem`, and others
+- `ocifs`_ for access to Oracle Cloud Object Storage, with protocol "oci://"
 - `ocilake`_ for OCI Data Lake storage
 - `ossfs`_ for Alibaba Cloud (Aliyun) Object Storage System (OSS)
 - `p9fs`_ for 9P (Plan 9 Filesystem Protocol) servers
-- `s3fs`_ for Amazon S3 and other compatible stores
+- `PyAthena`_ for S3 access to Amazon Athena, with protocol "s3://" or "s3a://"
+- `PyDrive2`_ for Google Drive access
+- `s3fs`_ for Amazon S3 and other compatible stores, with protocol "s3://"
+- `sshfs`_ for access to SSH servers, with protocol "ssh://" or "sftp://"
+- `swiftspec`_ for OpenStack SWIFT, with protocol "swift://"
+- `tosfs`_ for ByteDance volcano engine Tinder Object Storage (TOS)
 - `wandbfs`_ to access Wandb run data (experimental)
-- `webdav4`_ for WebDAV
+- `wandbfsspec`_ to access Weights & Biases (experimental)
+- `webdav4`_ for WebDAV, with protocol "webdav://" or "dav://"
 - `xrootd`_ for xrootd, with protocol "root://"
 
 .. _abfs: https://github.com/dask/adlfs
 .. _adl: https://github.com/dask/adlfs
 .. _alluxiofs: https://github.com/fsspec/alluxiofs
 .. _boxfs: https://github.com/IBM/boxfs
-.. _dropbox: https://github.com/MarineChap/intake_dropbox
+.. _csvbase: https://github.com/calpaterson/csvbase-client
+.. _dropbox: https://github.com/fsspec/dropboxdrivefs
 .. _dvc: https://github.com/iterative/dvc
+.. _fsspec-encrypted: https://github.com/thevgergroup/fsspec-encrypted
 .. _gcsfs: https://gcsfs.readthedocs.io/en/latest/
 .. _gdrive: https://github.com/fsspec/gdrivefs
+.. _git: https://github.com/iterative/scmrepo
+.. _hdfs-native: https://github.com/Kimahriman/hdfs-native/blob/master/python/hdfs_native/fsspec.py
+.. _httpfs-sync: https://github.com/moradology/httpfs-sync
 .. _huggingface_hub: https://huggingface.co/docs/huggingface_hub/main/en/guides/hf_file_system
-.. _lakefs: https://github.com/appliedAI-Initiative/lakefs-spec
-.. _ocifs: https://pypi.org/project/ocifs
+.. _ipfsspec: https://github.com/fsspec/ipfsspec
+.. _irods: https://github.com/xwcl/irods_fsspec
+.. _lakefs: https://github.com/aai-institute/lakefs-spec
+.. _morefs: https://github.com/iterative/morefs
+.. _ocifs: https://ocifs.readthedocs.io/en/latest/
 .. _ocilake: https://github.com/oracle/ocifs
 .. _ossfs: https://github.com/fsspec/ossfs
 .. _p9fs: https://github.com/pbchekin/p9fs-py
+.. _PyAthena: https://github.com/laughingman7743/PyAthena
+.. _PyDrive2: https://github.com/iterative/PyDrive2
 .. _s3fs: https://s3fs.readthedocs.io/en/latest/
+.. _sshfs: https://github.com/fsspec/sshfs
+.. _swiftspec: https://github.com/fsspec/swiftspec
+.. _tosfs: https://tosfs.readthedocs.io/en/latest/
 .. _wandbfs: https://github.com/jkulhanek/wandbfs
+.. _wandbfsspec: https://github.com/alvarobartt/wandbfsspec
 .. _webdav4: https://github.com/skshetry/webdav4
 .. _xrootd: https://github.com/CoffeaTeam/fsspec-xrootd
 
diff --git a/docs/source/async.rst b/docs/source/async.rst
index 58334b3..dc44df3 100644
--- a/docs/source/async.rst
+++ b/docs/source/async.rst
@@ -152,3 +152,37 @@ available as the attribute ``.loop``.
 
     <script data-goatcounter="https://fsspec.goatcounter.com/count"
         async src="//gc.zgo.at/count.js"></script>
+
+AsyncFileSystemWrapper
+----------------------
+
+The `AsyncFileSystemWrapper` class is an experimental feature that allows you to convert
+a synchronous filesystem into an asynchronous one. This is useful for quickly integrating
+synchronous filesystems into workflows that may expect `AsyncFileSystem` instances.
+
+Basic Usage
+~~~~~~~~~~~
+
+To use `AsyncFileSystemWrapper`, wrap any synchronous filesystem to work in an asynchronous context.
+In this example, the synchronous `LocalFileSystem` is wrapped, creating an `AsyncFileSystem` instance
+backed by the normal, synchronous methods of `LocalFileSystem`:
+
+.. code-block:: python
+
+    import asyncio
+    import fsspec
+    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+
+    async def async_copy_file():
+        sync_fs = fsspec.filesystem('file')  # by-default synchronous, local filesystem
+        async_fs = AsyncFileSystemWrapper(sync_fs)
+        return await async_fs._copy('/source/file.txt', '/destination/file.txt')
+
+    asyncio.run(async_copy_file())
+
+Limitations
+-----------
+
+This is experimental. Users should not expect this wrapper to magically make things faster.
+It is primarily provided to allow usage of synchronous filesystems with interfaces that expect
+`AsyncFileSystem` instances.
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index b19ee9e..fea257b 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,27 @@
 Changelog
 =========
 
+2024.10.0
+---------
+
+Fixes
+
+- Performance of memoryFS rm (#1725)
+- Performance of git FS info (#1712)
+- Avoid git hex for newer pygit (#1703)
+- tests fix for zip (#1700, 1691)
+- missing open_async for dirFS (#1698)
+- handle pathlib in zip (#1689)
+- skip tests needing kerchunk if not installed (#1689)
+- allow repeated kwargs in unchain (#1673)
+
+Other
+
+- Code style (#1704, 1706)
+- allow pyarrow in referenceFS parquet (#1692)
+- don't hardcode test port for parallel runs (#1690)
+
+
 2024.9.0
 --------
 
diff --git a/fsspec/caching.py b/fsspec/caching.py
index a3f7a1c..bc74ad2 100644
--- a/fsspec/caching.py
+++ b/fsspec/caching.py
@@ -7,6 +7,8 @@ import math
 import os
 import threading
 import warnings
+from itertools import groupby
+from operator import itemgetter
 from concurrent.futures import Future, ThreadPoolExecutor
 from typing import (
     TYPE_CHECKING,
@@ -161,21 +163,39 @@ class MMapCache(BaseCache):
             return b""
         start_block = start // self.blocksize
         end_block = end // self.blocksize
-        need = [i for i in range(start_block, end_block + 1) if i not in self.blocks]
-        hits = [i for i in range(start_block, end_block + 1) if i in self.blocks]
-        self.miss_count += len(need)
-        self.hit_count += len(hits)
-        while need:
-            # TODO: not a for loop so we can consolidate blocks later to
-            # make fewer fetch calls; this could be parallel
-            i = need.pop(0)
-
-            sstart = i * self.blocksize
-            send = min(sstart + self.blocksize, self.size)
+        block_range = range(start_block, end_block + 1)
+        # Determine which blocks need to be fetched. This sequence is sorted by construction.
+        need = (i for i in block_range if i not in self.blocks)
+        # Count the number of blocks already cached
+        self.hit_count += sum(1 for i in block_range if i in self.blocks)
+
+        # Consolidate needed blocks.
+        # Algorithm adapted from Python 2.x itertools documentation.
+        # We are grouping an enumerated sequence of blocks. By comparing when the difference
+        # between an ascending range (provided by enumerate) and the needed block numbers
+        # we can detect when the block number skips values. The key computes this difference.
+        # Whenever the difference changes, we know that we have previously cached block(s),
+        # and a new group is started. In other words, this algorithm neatly groups
+        # runs of consecutive block numbers so they can be fetched together.
+        for _, _blocks in groupby(enumerate(need), key=lambda x: x[0] - x[1]):
+            # Extract the blocks from the enumerated sequence
+            _blocks = tuple(map(itemgetter(1), _blocks))
+            # Compute start of first block
+            sstart = _blocks[0] * self.blocksize
+            # Compute the end of the last block. Last block may not be full size.
+            send = min(_blocks[-1] * self.blocksize + self.blocksize, self.size)
+
+            # Fetch bytes (could be multiple consecutive blocks)
             self.total_requested_bytes += send - sstart
-            logger.debug(f"MMap get block #{i} ({sstart}-{send})")
+            logger.debug(
+                f"MMap get blocks {_blocks[0]}-{_blocks[-1]} ({sstart}-{send})"
+            )
             self.cache[sstart:send] = self.fetcher(sstart, send)
-            self.blocks.add(i)
+
+            # Update set of cached blocks
+            self.blocks.update(_blocks)
+            # Update cache statistics with number of blocks we had to cache
+            self.miss_count += len(_blocks)
 
         return self.cache[start:end]
 
diff --git a/fsspec/core.py b/fsspec/core.py
index e79f2a8..1d3c04c 100644
--- a/fsspec/core.py
+++ b/fsspec/core.py
@@ -329,12 +329,19 @@ def open_files(
 
 
 def _un_chain(path, kwargs):
-    x = re.compile(".*[^a-z]+.*")  # test for non protocol-like single word
-    bits = (
-        [p if "://" in p or x.match(p) else p + "://" for p in path.split("::")]
-        if "::" in path
-        else [path]
-    )
+    # Avoid a circular import
+    from fsspec.implementations.cached import CachingFileSystem
+
+    if "::" in path:
+        x = re.compile(".*[^a-z]+.*")  # test for non protocol-like single word
+        bits = []
+        for p in path.split("::"):
+            if "://" in p or x.match(p):
+                bits.append(p)
+            else:
+                bits.append(p + "://")
+    else:
+        bits = [path]
     # [[url, protocol, kwargs], ...]
     out = []
     previous_bit = None
@@ -346,12 +353,12 @@ def _un_chain(path, kwargs):
         kws = kwargs.pop(protocol, {})
         if bit is bits[0]:
             kws.update(kwargs)
-        kw = dict(**extra_kwargs, **kws)
+        kw = dict(
+            **{k: v for k, v in extra_kwargs.items() if k not in kws or v != kws[k]},
+            **kws,
+        )
         bit = cls._strip_protocol(bit)
-        if (
-            protocol in {"blockcache", "filecache", "simplecache"}
-            and "target_protocol" not in kw
-        ):
+        if "target_protocol" not in kw and issubclass(cls, CachingFileSystem):
             bit = previous_bit
         out.append((bit, protocol, kw))
         previous_bit = bit
@@ -673,9 +680,7 @@ def get_fs_token_paths(
         elif not isinstance(paths, list):
             paths = list(paths)
     else:
-        if "w" in mode and expand:
-            paths = _expand_paths(paths, name_function, num)
-        elif "x" in mode and expand:
+        if ("w" in mode or "x" in mode) and expand:
             paths = _expand_paths(paths, name_function, num)
         elif "*" in paths:
             paths = [f for f in sorted(fs.glob(paths)) if not fs.isdir(f)]
diff --git a/fsspec/implementations/asyn_wrapper.py b/fsspec/implementations/asyn_wrapper.py
new file mode 100644
index 0000000..9ba7811
--- /dev/null
+++ b/fsspec/implementations/asyn_wrapper.py
@@ -0,0 +1,96 @@
+import asyncio
+import inspect
+import functools
+from fsspec.asyn import AsyncFileSystem
+
+
+def async_wrapper(func, obj=None):
+    """
+    Wraps a synchronous function to make it awaitable.
+
+    Parameters
+    ----------
+    func : callable
+        The synchronous function to wrap.
+    obj : object, optional
+        The instance to bind the function to, if applicable.
+
+    Returns
+    -------
+    coroutine
+        An awaitable version of the function.
+    """
+
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        return await asyncio.to_thread(func, *args, **kwargs)
+
+    return wrapper
+
+
+class AsyncFileSystemWrapper(AsyncFileSystem):
+    """
+    A wrapper class to convert a synchronous filesystem into an asynchronous one.
+
+    This class takes an existing synchronous filesystem implementation and wraps all
+    its methods to provide an asynchronous interface.
+
+    Parameters
+    ----------
+    sync_fs : AbstractFileSystem
+        The synchronous filesystem instance to wrap.
+    """
+
+    def __init__(self, sync_fs, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.asynchronous = True
+        self.fs = sync_fs
+        self._wrap_all_sync_methods()
+
+    @property
+    def fsid(self):
+        return f"async_{self.fs.fsid}"
+
+    def _wrap_all_sync_methods(self):
+        """
+        Wrap all synchronous methods of the underlying filesystem with asynchronous versions.
+        """
+        for method_name in dir(self.fs):
+            if method_name.startswith("_"):
+                continue
+
+            attr = inspect.getattr_static(self.fs, method_name)
+            if isinstance(attr, property):
+                continue
+
+            method = getattr(self.fs, method_name)
+            if callable(method) and not asyncio.iscoroutinefunction(method):
+                async_method = async_wrapper(method, obj=self)
+                setattr(self, f"_{method_name}", async_method)
+
+    @classmethod
+    def wrap_class(cls, sync_fs_class):
+        """
+        Create a new class that can be used to instantiate an AsyncFileSystemWrapper
+        with lazy instantiation of the underlying synchronous filesystem.
+
+        Parameters
+        ----------
+        sync_fs_class : type
+            The class of the synchronous filesystem to wrap.
+
+        Returns
+        -------
+        type
+            A new class that wraps the provided synchronous filesystem class.
+        """
+
+        class GeneratedAsyncFileSystemWrapper(cls):
+            def __init__(self, *args, **kwargs):
+                sync_fs = sync_fs_class(*args, **kwargs)
+                super().__init__(sync_fs)
+
+        GeneratedAsyncFileSystemWrapper.__name__ = (
+            f"Async{sync_fs_class.__name__}Wrapper"
+        )
+        return GeneratedAsyncFileSystemWrapper
diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 447e4f2..8b8b832 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -612,7 +612,7 @@ class WholeFileCacheFileSystem(CachingFileSystem):
         **kwargs,
     ):
         paths = self.expand_path(
-            path, recursive=recursive, maxdepth=kwargs.get("maxdepth", None)
+            path, recursive=recursive, maxdepth=kwargs.get("maxdepth")
         )
         getpaths = []
         storepaths = []
diff --git a/fsspec/implementations/ftp.py b/fsspec/implementations/ftp.py
index f3471b9..4186b6c 100644
--- a/fsspec/implementations/ftp.py
+++ b/fsspec/implementations/ftp.py
@@ -387,7 +387,7 @@ def _mlsd2(ftp, path="."):
                 "size": split_line[4],
             },
         )
-        if "d" == this[1]["unix.mode"][0]:
+        if this[1]["unix.mode"][0] == "d":
             this[1]["type"] = "dir"
         else:
             this[1]["type"] = "file"
diff --git a/fsspec/implementations/git.py b/fsspec/implementations/git.py
index fde671b..7b9d353 100644
--- a/fsspec/implementations/git.py
+++ b/fsspec/implementations/git.py
@@ -55,6 +55,8 @@ class GitFileSystem(AbstractFileSystem):
         tree = comm.tree
         for part in parts:
             if part and isinstance(tree, pygit2.Tree):
+                if part not in tree:
+                    raise FileNotFoundError(path)
                 tree = tree[part]
         return tree
 
@@ -69,46 +71,32 @@ class GitFileSystem(AbstractFileSystem):
             out["ref"], path = path.split("@", 1)
         return out
 
+    @staticmethod
+    def _object_to_info(obj, path=None):
+        # obj.name and obj.filemode are None for the root tree!
+        is_dir = isinstance(obj, pygit2.Tree)
+        return {
+            "type": "directory" if is_dir else "file",
+            "name": (
+                "/".join([path, obj.name or ""]).lstrip("/") if path else obj.name
+            ),
+            "hex": str(obj.id),
+            "mode": "100644" if obj.filemode is None else f"{obj.filemode:o}",
+            "size": 0 if is_dir else obj.size,
+        }
+
     def ls(self, path, detail=True, ref=None, **kwargs):
-        path = self._strip_protocol(path)
-        tree = self._path_to_object(path, ref)
-        if isinstance(tree, pygit2.Tree):
-            out = []
-            for obj in tree:
-                if isinstance(obj, pygit2.Tree):
-                    out.append(
-                        {
-                            "type": "directory",
-                            "name": "/".join([path, obj.name]).lstrip("/"),
-                            "hex": str(obj.id),
-                            "mode": f"{obj.filemode:o}",
-                            "size": 0,
-                        }
-                    )
-                else:
-                    out.append(
-                        {
-                            "type": "file",
-                            "name": "/".join([path, obj.name]).lstrip("/"),
-                            "hex": str(obj.id),
-                            "mode": f"{obj.filemode:o}",
-                            "size": obj.size,
-                        }
-                    )
-        else:
-            obj = tree
-            out = [
-                {
-                    "type": "file",
-                    "name": obj.name,
-                    "hex": str(obj.id),
-                    "mode": f"{obj.filemode:o}",
-                    "size": obj.size,
-                }
-            ]
-        if detail:
-            return out
-        return [o["name"] for o in out]
+        tree = self._path_to_object(self._strip_protocol(path), ref)
+        return [
+            GitFileSystem._object_to_info(obj, path)
+            if detail
+            else GitFileSystem._object_to_info(obj, path)["name"]
+            for obj in (tree if isinstance(tree, pygit2.Tree) else [tree])
+        ]
+
+    def info(self, path, ref=None, **kwargs):
+        tree = self._path_to_object(self._strip_protocol(path), ref)
+        return GitFileSystem._object_to_info(tree, path)
 
     def ukey(self, path, ref=None):
         return self.info(path, ref=ref)["hex"]
diff --git a/fsspec/implementations/local.py b/fsspec/implementations/local.py
index f032d8a..4c58823 100644
--- a/fsspec/implementations/local.py
+++ b/fsspec/implementations/local.py
@@ -60,7 +60,12 @@ class LocalFileSystem(AbstractFileSystem):
         info = self.info(path)
         if info["type"] == "directory":
             with os.scandir(path) as it:
-                infos = [self.info(f) for f in it]
+                infos = []
+                for f in it:
+                    try:
+                        infos.append(self.info(f))
+                    except FileNotFoundError:
+                        pass
         else:
             infos = [info]
 
diff --git a/fsspec/implementations/memory.py b/fsspec/implementations/memory.py
index 93860af..3b6f0d2 100644
--- a/fsspec/implementations/memory.py
+++ b/fsspec/implementations/memory.py
@@ -248,6 +248,10 @@ class MemoryFileSystem(AbstractFileSystem):
         except KeyError as e:
             raise FileNotFoundError(path) from e
 
+    def isfile(self, path):
+        path = self._strip_protocol(path)
+        return path in self.store
+
     def rm(self, path, recursive=False, maxdepth=None):
         if isinstance(path, str):
             path = self._strip_protocol(path)
@@ -255,14 +259,14 @@ class MemoryFileSystem(AbstractFileSystem):
             path = [self._strip_protocol(p) for p in path]
         paths = self.expand_path(path, recursive=recursive, maxdepth=maxdepth)
         for p in reversed(paths):
+            if self.isfile(p):
+                self.rm_file(p)
             # If the expanded path doesn't exist, it is only because the expanded
             # path was a directory that does not exist in self.pseudo_dirs. This
             # is possible if you directly create files without making the
             # directories first.
-            if not self.exists(p):
+            elif not self.exists(p):
                 continue
-            if self.isfile(p):
-                self.rm_file(p)
             else:
                 self.rmdir(p)
 
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index d5d3f29..d8df6ee 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -5,8 +5,8 @@ import itertools
 import logging
 import math
 import os
-from itertools import chain
 from functools import lru_cache
+from itertools import chain
 from typing import TYPE_CHECKING, Literal
 
 import fsspec.core
@@ -41,7 +41,7 @@ def _first(d):
 
 def _prot_in_references(path, references):
     ref = references.get(path)
-    if isinstance(ref, (list, tuple)):
+    if isinstance(ref, (list, tuple)) and isinstance(ref[0], str):
         return split_protocol(ref[0])[0] if ref[0] else ref[0]
 
 
@@ -173,8 +173,11 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
             """cached parquet file loader"""
             path = self.url.format(field=field, record=record)
             data = io.BytesIO(self.fs.cat_file(path))
-            df = self.pd.read_parquet(data, engine=self.engine)
-            refs = {c: df[c].to_numpy() for c in df.columns}
+            try:
+                df = self.pd.read_parquet(data, engine=self.engine)
+                refs = {c: df[c].to_numpy() for c in df.columns}
+            except IOError:
+                refs = None
             return refs
 
         self.open_refs = open_refs
@@ -803,7 +806,9 @@ class ReferenceFileSystem(AsyncFileSystem):
             return part_or_url[start:end]
         protocol, _ = split_protocol(part_or_url)
         try:
-            await self.fss[protocol]._cat_file(part_or_url, start=start, end=end)
+            return await self.fss[protocol]._cat_file(
+                part_or_url, start=start0, end=end0
+            )
         except Exception as e:
             raise ReferenceNotReachable(path, part_or_url) from e
 
@@ -871,6 +876,9 @@ class ReferenceFileSystem(AsyncFileSystem):
                 # found and on_error is "raise"
                 try:
                     u, s, e = self._cat_common(p)
+                    if not isinstance(u, (bytes, str)):
+                        # nan/None from parquet
+                        continue
                 except FileNotFoundError as err:
                     if on_error == "raise":
                         raise
diff --git a/fsspec/mapping.py b/fsspec/mapping.py
index 8fb9b9e..752eef3 100644
--- a/fsspec/mapping.py
+++ b/fsspec/mapping.py
@@ -112,7 +112,7 @@ class FSMap(MutableMapping):
             for k, v in out.items()
         }
         return {
-            key: out[k2]
+            key: out[k2] if on_error == "raise" else out.get(k2, KeyError(k2))
             for key, k2 in zip(keys, keys2)
             if on_error == "return" or not isinstance(out[k2], BaseException)
         }
diff --git a/fsspec/parquet.py b/fsspec/parquet.py
index 5a0fb95..faedb7b 100644
--- a/fsspec/parquet.py
+++ b/fsspec/parquet.py
@@ -336,7 +336,7 @@ def _add_header_magic(data):
     # Add b"PAR1" to file headers
     for path in list(data.keys()):
         add_magic = True
-        for k in data[path].keys():
+        for k in data[path]:
             if k[0] == 0 and k[1] >= 4:
                 add_magic = False
                 break
diff --git a/fsspec/registry.py b/fsspec/registry.py
index c261b9b..02094ea 100644
--- a/fsspec/registry.py
+++ b/fsspec/registry.py
@@ -202,6 +202,10 @@ known_implementations = {
         "err": 'SFTPFileSystem requires "paramiko" to be installed',
     },
     "tar": {"class": "fsspec.implementations.tar.TarFileSystem"},
+    "tosfs": {
+        "class": "tosfs.TosFileSystem",
+        "err": "Install tosfs to access ByteDance volcano engine Tinder Object Storage",
+    },
     "wandb": {"class": "wandbfs.WandbFS", "err": "Install wandbfs to access wandb"},
     "webdav": {
         "class": "webdav4.fsspec.WebdavFileSystem",
diff --git a/fsspec/spec.py b/fsspec/spec.py
index e67d280..9659f2e 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -408,7 +408,7 @@ class AbstractFileSystem(metaclass=_Cached):
         topdown: bool (True)
             Whether to walk the directory tree from the top downwards or from
             the bottom upwards.
-        on_error: "omit", "raise", a collable
+        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
@@ -648,7 +648,7 @@ class AbstractFileSystem(metaclass=_Cached):
         Returns a single dictionary, with exactly the same information as ``ls``
         would with ``detail=True``.
 
-        The default implementation should calls ls and could be overridden by a
+        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
@@ -1915,7 +1915,7 @@ class AbstractBufferedFile(io.IOBase):
 
     def _fetch_range(self, start, end):
         """Get the specified set of bytes from remote"""
-        raise NotImplementedError
+        return self.fs.cat_file(self.path, start=start, end=end)
 
     def read(self, length=-1):
         """
diff --git a/pyproject.toml b/pyproject.toml
index 4836871..75a70d4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,6 +21,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
 ]
 
 [project.optional-dependencies]
@@ -162,6 +163,7 @@ select = [
     "E9",
     "F",
     "FURB",
+    "I",
     "LOG",
     "PERF",
     "PLC",
@@ -180,8 +182,8 @@ select = [
     "RUF006",
     "RUF015",
     "RUF024",
+    "SIM",
     "SLOT",
-    "SIM101",
 ]
 ignore = [
 	# Loop control variable `loop` not used within loop body
@@ -206,6 +208,12 @@ ignore = [
     # Fix these codes later
     "G004",
     "PERF203",
+    "SIM102",
+    "SIM105",
+    "SIM108",
+    "SIM114",
+    "SIM115",
+    "SIM117",
     # https://github.com/astral-sh/ruff/issues/7871
     "UP038",
     # https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules


Test patch:
diff --git a/fsspec/implementations/tests/test_asyn_wrapper.py b/fsspec/implementations/tests/test_asyn_wrapper.py
new file mode 100644
index 0000000..d292020
--- /dev/null
+++ b/fsspec/implementations/tests/test_asyn_wrapper.py
@@ -0,0 +1,142 @@
+import asyncio
+import pytest
+import os
+
+import fsspec
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+from fsspec.implementations.local import LocalFileSystem
+from .test_local import csv_files, filetexts
+
+
+def test_is_async():
+    fs = fsspec.filesystem("file")
+    async_fs = AsyncFileSystemWrapper(fs)
+    assert async_fs.async_impl
+
+
+def test_class_wrapper():
+    fs_cls = LocalFileSystem
+    async_fs_cls = AsyncFileSystemWrapper.wrap_class(fs_cls)
+    assert async_fs_cls.__name__ == "AsyncLocalFileSystemWrapper"
+    async_fs = async_fs_cls()
+    assert async_fs.async_impl
+
+
+@pytest.mark.asyncio
+async def test_cats():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        result = await async_fs._cat(".test.fakedata.1.csv")
+        assert result == b"a,b\n1,2\n"
+
+        out = set(
+            (
+                await async_fs._cat([".test.fakedata.1.csv", ".test.fakedata.2.csv"])
+            ).values()
+        )
+        assert out == {b"a,b\n1,2\n", b"a,b\n3,4\n"}
+
+        result = await async_fs._cat(".test.fakedata.1.csv", None, None)
+        assert result == b"a,b\n1,2\n"
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=1, end=6)
+        assert result == b"a,b\n1,2\n"[1:6]
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=-1)
+        assert result == b"a,b\n1,2\n"[-1:]
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=1, end=-2)
+        assert result == b"a,b\n1,2\n"[1:-2]
+
+        # test synchronous API is available as expected
+        result = async_fs.cat(".test.fakedata.1.csv", start=1, end=-2)
+        assert result == b"a,b\n1,2\n"[1:-2]
+
+        out = set(
+            (
+                await async_fs._cat(
+                    [".test.fakedata.1.csv", ".test.fakedata.2.csv"], start=1, end=-1
+                )
+            ).values()
+        )
+        assert out == {b"a,b\n1,2\n"[1:-1], b"a,b\n3,4\n"[1:-1]}
+
+
+@pytest.mark.asyncio
+async def test_basic_crud_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._touch(".test.fakedata.3.csv")
+        assert await async_fs._exists(".test.fakedata.3.csv")
+
+        data = await async_fs._cat(".test.fakedata.1.csv")
+        assert data == b"a,b\n1,2\n"
+
+        await async_fs._pipe(".test.fakedata.1.csv", b"a,b\n5,6\n")
+        data = await async_fs._cat(".test.fakedata.1.csv")
+        assert data == b"a,b\n5,6\n"
+
+        await async_fs._rm(".test.fakedata.1.csv")
+        assert not await async_fs._exists(".test.fakedata.1.csv")
+
+
+@pytest.mark.asyncio
+async def test_error_handling():
+    fs = fsspec.filesystem("file")
+    async_fs = AsyncFileSystemWrapper(fs)
+
+    with pytest.raises(FileNotFoundError):
+        await async_fs._cat(".test.non_existent.csv")
+
+    with pytest.raises(FileNotFoundError):
+        await async_fs._rm(".test.non_existent.csv")
+
+
+@pytest.mark.asyncio
+async def test_concurrent_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        async def read_file(file_path):
+            return await async_fs._cat(file_path)
+
+        results = await asyncio.gather(
+            read_file(".test.fakedata.1.csv"),
+            read_file(".test.fakedata.2.csv"),
+            read_file(".test.fakedata.1.csv"),
+        )
+
+        assert results == [b"a,b\n1,2\n", b"a,b\n3,4\n", b"a,b\n1,2\n"]
+
+
+@pytest.mark.asyncio
+async def test_directory_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._makedirs("new_directory")
+        assert await async_fs._isdir("new_directory")
+
+        files = await async_fs._ls(".")
+        filenames = [os.path.basename(file) for file in files]
+
+        assert ".test.fakedata.1.csv" in filenames
+        assert ".test.fakedata.2.csv" in filenames
+        assert "new_directory" in filenames
+
+
+@pytest.mark.asyncio
+async def test_batch_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._rm([".test.fakedata.1.csv", ".test.fakedata.2.csv"])
+        assert not await async_fs._exists(".test.fakedata.1.csv")
+        assert not await async_fs._exists(".test.fakedata.2.csv")
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 67edbc8..c9222d5 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -267,7 +267,7 @@ def test_blockcache_workflow(ftp_writable, tmp_path, force_save_pickle):
         assert f.read(5) == b"test\n"
 
 
-@pytest.mark.parametrize("impl", ["filecache", "blockcache"])
+@pytest.mark.parametrize("impl", ["filecache", "blockcache", "cached"])
 def test_workflow(ftp_writable, impl):
     host, port, user, pw = ftp_writable
     fs = FTPFileSystem(host, port, user, pw)
@@ -295,7 +295,7 @@ def test_workflow(ftp_writable, impl):
         )  # new value, because we overwrote the cached location
 
 
-@pytest.mark.parametrize("impl", ["simplecache", "blockcache"])
+@pytest.mark.parametrize("impl", ["simplecache", "blockcache", "cached"])
 def test_glob(ftp_writable, impl):
     host, port, user, pw = ftp_writable
     fs = FTPFileSystem(host, port, user, pw)
@@ -622,7 +622,7 @@ def test_metadata_save_blocked(ftp_writable, caplog):
     assert "Cache save failed due to interpreter shutdown" in caplog.text
 
 
-@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache"])
+@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache", "cached"])
 def test_local_filecache_creates_dir_if_needed(impl):
     import tempfile
 
@@ -875,7 +875,7 @@ def test_filecache_with_checks():
     assert fs.cat(f1) == data * 2  # changed, since origin changed
 
 
-@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache"])
+@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache", "cached"])
 @pytest.mark.parametrize("fs", ["local", "multi"], indirect=["fs"])
 def test_filecache_takes_fs_instance(impl, fs):
     origin = tempfile.mkdtemp()
@@ -889,7 +889,7 @@ def test_filecache_takes_fs_instance(impl, fs):
     assert fs2.cat(f1) == data
 
 
-@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache"])
+@pytest.mark.parametrize("impl", ["filecache", "simplecache", "blockcache", "cached"])
 @pytest.mark.parametrize("fs", ["local", "multi"], indirect=["fs"])
 def test_filecache_serialization(impl, fs):
     fs1 = fsspec.filesystem(impl, fs=fs)
@@ -1031,7 +1031,9 @@ def test_multi_cache(protocol):
             assert f.read() == b"hello"
 
 
-@pytest.mark.parametrize("protocol", ["simplecache", "filecache", "blockcache"])
+@pytest.mark.parametrize(
+    "protocol", ["simplecache", "filecache", "blockcache", "cached"]
+)
 def test_multi_cat(protocol, ftp_writable):
     host, port, user, pw = ftp_writable
     fs = FTPFileSystem(host, port, user, pw)
@@ -1064,7 +1066,9 @@ def test_multi_cache_chain(protocol):
         assert files[0].read() == b"hello"
 
 
-@pytest.mark.parametrize("protocol", ["blockcache", "simplecache", "filecache"])
+@pytest.mark.parametrize(
+    "protocol", ["blockcache", "cached", "simplecache", "filecache"]
+)
 def test_strip(protocol):
     fs = fsspec.filesystem(protocol, target_protocol="memory")
     url1 = "memory://afile"
@@ -1235,9 +1239,11 @@ def test_cache_dir_auto_deleted(temp_cache, tmpdir):
         assert local.exists(cache_dir)
 
 
-@pytest.mark.parametrize("protocol", ["filecache", "blockcache", "simplecache"])
+@pytest.mark.parametrize(
+    "protocol", ["filecache", "blockcache", "cached", "simplecache"]
+)
 def test_cache_size(tmpdir, protocol):
-    if win and protocol == "blockcache":
+    if win and protocol in {"blockcache", "cached"}:
         pytest.skip("Windows file locking affects blockcache size tests")
 
     source = os.path.join(tmpdir, "source")
diff --git a/fsspec/implementations/tests/test_git.py b/fsspec/implementations/tests/test_git.py
index ffa7b47..2aeb544 100644
--- a/fsspec/implementations/tests/test_git.py
+++ b/fsspec/implementations/tests/test_git.py
@@ -61,6 +61,39 @@ def test_refs(repo):
         assert f.read() == b"data3"
 
 
+def _check_FileNotFoundError(f, *args, **kwargs):
+    with pytest.raises(FileNotFoundError):
+        f(*args, **kwargs)
+
+
+def test_file_existence_checks(repo):
+    d, sha = repo
+
+    fs, _ = fsspec.url_to_fs(f"git://{d}:abranch@")
+
+    assert fs.lexists("inner")
+    assert fs.exists("inner")
+    assert fs.isdir("inner")
+    assert fs.info("inner")
+    assert fs.ls("inner")
+
+    assert fs.lexists("inner/file1")
+    assert fs.exists("inner/file1")
+    assert fs.info("inner/file1")
+    assert fs.ls("inner/file1")
+
+    assert not fs.lexists("non-existing-file")
+    assert not fs.exists("non-existing-file")
+
+    assert not fs.isfile("non-existing-file")
+    assert not fs.isdir("non-existing-file")
+
+    _check_FileNotFoundError(fs.info, "non-existing-file")
+    _check_FileNotFoundError(fs.size, "non-existing-file")
+    _check_FileNotFoundError(fs.ls, "non-existing-file")
+    _check_FileNotFoundError(fs.open, "non-existing-file")
+
+
 def test_url(repo):
     d, sha = repo
     fs, _, paths = fsspec.core.get_fs_token_paths(f"git://file1::file://{d}")
diff --git a/fsspec/implementations/tests/test_local.py b/fsspec/implementations/tests/test_local.py
index ef39279..15cd754 100644
--- a/fsspec/implementations/tests/test_local.py
+++ b/fsspec/implementations/tests/test_local.py
@@ -247,7 +247,7 @@ def test_not_found():
 def test_isfile():
     fs = LocalFileSystem()
     with filetexts(files, mode="b"):
-        for f in files.keys():
+        for f in files:
             assert fs.isfile(f)
             assert fs.isfile(f"file://{f}")
         assert not fs.isfile("not-a-file")
@@ -257,7 +257,7 @@ def test_isfile():
 def test_isdir():
     fs = LocalFileSystem()
     with filetexts(files, mode="b"):
-        for f in files.keys():
+        for f in files:
             assert fs.isdir(os.path.dirname(os.path.abspath(f)))
             assert not fs.isdir(f)
         assert not fs.isdir("not-a-dir")
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index d82dc17..0063d11 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -513,6 +513,35 @@ def test_cat_file_ranges(m):
     assert fs.cat_file("d", 1, -3) == other[4:10][1:-3]
 
 
+@pytest.mark.asyncio
+async def test_async_cat_file_ranges():
+    fsspec.get_filesystem_class("http").clear_instance_cache()
+    fss = fsspec.filesystem("https", asynchronous=True)
+    session = await fss.set_session()
+
+    fs = fsspec.filesystem(
+        "reference",
+        fo={
+            "version": 1,
+            "refs": {
+                "reference_time/0": [
+                    "https://noaa-nwm-retro-v2-0-pds.s3.amazonaws.com/full_physics/2017/201704010000.CHRTOUT_DOMAIN1.comp",
+                    39783,
+                    12,
+                ],
+            },
+        },
+        fs={"https": fss},
+        remote_protocol="https",
+        asynchronous=True,
+    )
+
+    assert (
+        await fs._cat_file("reference_time/0") == b"x^K0\xa9d\x04\x00\x03\x13\x01\x0f"
+    )
+    await session.close()
+
+
 @pytest.mark.parametrize(
     "fo",
     [
@@ -792,3 +821,44 @@ def test_deep_parq(m, engine):
         "instant/one/.zarray",
         "instant/one/0",
     ]
+
+
+def test_parquet_no_data(m):
+    zarr = pytest.importorskip("zarr")
+    lz = fsspec.implementations.reference.LazyReferenceMapper.create(
+        "memory://out.parq", fs=m
+    )
+
+    g = zarr.open_group(lz, mode="w")
+    arr = g.create_dataset(
+        name="one",
+        dtype="int32",
+        shape=(10,),
+        chunks=(5,),
+        compression=None,
+        fill_value=1,
+    )
+    lz.flush()
+
+    assert (arr[:] == 1).all()
+
+
+def test_parquet_no_references(m):
+    zarr = pytest.importorskip("zarr")
+    lz = fsspec.implementations.reference.LazyReferenceMapper.create(
+        "memory://out.parq", fs=m
+    )
+
+    g = zarr.open_group(lz, mode="w")
+    arr = g.create_dataset(
+        name="one",
+        dtype="int32",
+        shape=(),
+        chunks=(),
+        compression=None,
+        fill_value=1,
+    )
+    lz.flush()
+    arr[...]
+
+    assert arr[...].tolist() == 1  #  scalar, equal to fill value
diff --git a/fsspec/implementations/tests/test_zip.py b/fsspec/implementations/tests/test_zip.py
index 14d0008..8142737 100644
--- a/fsspec/implementations/tests/test_zip.py
+++ b/fsspec/implementations/tests/test_zip.py
@@ -162,7 +162,7 @@ def zip_file_fixture(tmp_path):
 
 
 def _assert_all_except_context_dependent_variables(result, expected_result):
-    for path in expected_result.keys():
+    for path in expected_result:
         assert result[path]
         fields = [
             "orig_filename",
diff --git a/fsspec/tests/test_core.py b/fsspec/tests/test_core.py
index 53592af..1cdeec9 100644
--- a/fsspec/tests/test_core.py
+++ b/fsspec/tests/test_core.py
@@ -465,3 +465,21 @@ def test_chained_url(ftp_writable):
 def test_automkdir_local():
     fs, _ = fsspec.core.url_to_fs("file://", auto_mkdir=True)
     assert fs.auto_mkdir is True
+
+
+def test_repeated_argument():
+    pytest.importorskip("adlfs")
+    from fsspec.core import url_to_fs
+
+    fs, url = url_to_fs(
+        "az://DIR@ACCOUNT.blob.core.windows.net/DATA",
+        anon=False,
+        account_name="ACCOUNT",
+    )
+    assert fs.storage_options == {"account_name": "ACCOUNT", "anon": False}
+    with pytest.raises(TypeError):
+        url_to_fs(
+            "az://DIR@ACCOUNT.blob.core.windows.net/DATA",
+            anon=False,
+            account_name="OTHER",
+        )
diff --git a/fsspec/tests/test_parquet.py b/fsspec/tests/test_parquet.py
index 5f38db7..7a76414 100644
--- a/fsspec/tests/test_parquet.py
+++ b/fsspec/tests/test_parquet.py
@@ -91,7 +91,7 @@ def test_open_parquet_file(
             # the footer metadata, so there should NOT
             # be a key for the last 8 bytes of the file
             bad_key = (file_size - 8, file_size)
-            assert bad_key not in data.keys()
+            assert bad_key not in data
 
         for (start, stop), byte_data in data.items():
             f.seek(start)