Repository: fsspec/filesystem_spec
Pull request #1749 (2024-11-13 21:43:26)

Make exclusive/atomic writes formally possible


Patch:
diff --git a/ci/environment-downstream.yml b/ci/environment-downstream.yml
index 7b02e61..d09f72d 100644
--- a/ci/environment-downstream.yml
+++ b/ci/environment-downstream.yml
@@ -5,3 +5,4 @@ dependencies:
   - python=3.11
   - pip:
       - git+https://github.com/dask/dask
+      - git+https://github.com/dask/dask-expr
diff --git a/docs/source/api.rst b/docs/source/api.rst
index 94c4cdd..5385acc 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -209,43 +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/fsspec/asyn.py b/fsspec/asyn.py
index de41839..adeb88f 100644
--- a/fsspec/asyn.py
+++ b/fsspec/asyn.py
@@ -408,7 +408,7 @@ class AsyncFileSystem(AbstractFileSystem):
                 continue
             raise ex
 
-    async def _pipe_file(self, path, value, **kwargs):
+    async def _pipe_file(self, path, value, mode="overwrite", **kwargs):
         raise NotImplementedError
 
     async def _pipe(self, path, value=None, batch_size=None, **kwargs):
@@ -517,7 +517,7 @@ class AsyncFileSystem(AbstractFileSystem):
             coros, batch_size=batch_size, nofiles=True, return_exceptions=True
         )
 
-    async def _put_file(self, lpath, rpath, **kwargs):
+    async def _put_file(self, lpath, rpath, mode="overwrite", **kwargs):
         raise NotImplementedError
 
     async def _put(
@@ -816,11 +816,9 @@ class AsyncFileSystem(AbstractFileSystem):
             p: info
             for p, info in sorted(allpaths.items())
             if pattern.match(
-                (
-                    p + "/"
-                    if append_slash_to_dirname and info["type"] == "directory"
-                    else p
-                )
+                p + "/"
+                if append_slash_to_dirname and info["type"] == "directory"
+                else p
             )
         }
 
diff --git a/fsspec/caching.py b/fsspec/caching.py
index bc74ad2..34231c8 100644
--- a/fsspec/caching.py
+++ b/fsspec/caching.py
@@ -7,9 +7,9 @@ import math
 import os
 import threading
 import warnings
+from concurrent.futures import Future, ThreadPoolExecutor
 from itertools import groupby
 from operator import itemgetter
-from concurrent.futures import Future, ThreadPoolExecutor
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -87,12 +87,7 @@ class BaseCache:
         if self.hit_count == 0 and self.miss_count == 0:
             # a cache that does nothing, this is for logs only
             return ""
-        return " , %s: %d hits, %d misses, %d total requested bytes" % (
-            self.name,
-            self.hit_count,
-            self.miss_count,
-            self.total_requested_bytes,
-        )
+        return f" , {self.name}: {self.hit_count} hits, {self.miss_count} misses, {self.total_requested_bytes} total requested bytes"
 
     def __repr__(self) -> str:
         # TODO: use rich for better formatting
diff --git a/fsspec/core.py b/fsspec/core.py
index 79580be..1d3c04c 100644
--- a/fsspec/core.py
+++ b/fsspec/core.py
@@ -680,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
index 9ba7811..0d46265 100644
--- a/fsspec/implementations/asyn_wrapper.py
+++ b/fsspec/implementations/asyn_wrapper.py
@@ -1,6 +1,7 @@
 import asyncio
-import inspect
 import functools
+import inspect
+
 from fsspec.asyn import AsyncFileSystem
 
 
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/http.py b/fsspec/implementations/http.py
index 47dfb88..ed1794f 100644
--- a/fsspec/implementations/http.py
+++ b/fsspec/implementations/http.py
@@ -273,8 +273,12 @@ class HTTPFileSystem(AsyncFileSystem):
         chunk_size=5 * 2**20,
         callback=DEFAULT_CALLBACK,
         method="post",
+        mode="overwrite",
         **kwargs,
     ):
+        if mode != "overwrite":
+            raise NotImplementedError("Exclusive write")
+
         async def gen_chunks():
             # Support passing arbitrary file-like objects
             # and use them instead of streams.
diff --git a/fsspec/implementations/memory.py b/fsspec/implementations/memory.py
index 3b6f0d2..c1c526b 100644
--- a/fsspec/implementations/memory.py
+++ b/fsspec/implementations/memory.py
@@ -126,12 +126,13 @@ class MemoryFileSystem(AbstractFileSystem):
             if not exist_ok:
                 raise
 
-    def pipe_file(self, path, value, **kwargs):
+    def pipe_file(self, path, value, mode="overwrite", **kwargs):
         """Set the bytes of given file
 
         Avoids copies of the data if possible
         """
-        self.open(path, "wb", data=value)
+        mode = "xb" if mode == "create" else "wb"
+        self.open(path, mode=mode, data=value)
 
     def rmdir(self, path):
         path = self._strip_protocol(path)
@@ -178,6 +179,8 @@ class MemoryFileSystem(AbstractFileSystem):
         **kwargs,
     ):
         path = self._strip_protocol(path)
+        if "x" in mode and self.exists(path):
+            raise FileExistsError
         if path in self.pseudo_dirs:
             raise IsADirectoryError(path)
         parent = path
@@ -197,7 +200,9 @@ class MemoryFileSystem(AbstractFileSystem):
                 return f
             else:
                 raise FileNotFoundError(path)
-        elif mode == "wb":
+        elif mode in {"wb", "xb"}:
+            if mode == "xb" and self.exists(path):
+                raise FileExistsError
             m = MemoryFile(self, path, kwargs.get("data"))
             if not self._intrans:
                 m.commit()
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index 79bff14..5c5372c 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
@@ -176,7 +176,7 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
             try:
                 df = self.pd.read_parquet(data, engine=self.engine)
                 refs = {c: df[c].to_numpy() for c in df.columns}
-            except IOError:
+            except OSError:
                 refs = None
             return refs
 
@@ -431,7 +431,7 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
         if len(partition) < self.record_size:
             try:
                 original = self.open_refs(field, record)
-            except IOError:
+            except OSError:
                 pass
 
         if original:
@@ -1181,13 +1181,17 @@ class ReferenceFileSystem(AsyncFileSystem):
         )  # ignores FileNotFound, just as well for directories
         self.dircache.clear()  # this is a bit heavy handed
 
-    async def _pipe_file(self, path, data):
+    async def _pipe_file(self, path, data, mode="overwrite", **kwargs):
+        if mode == "create" and self.exists(path):
+            raise FileExistsError
         # can be str or bytes
         self.references[path] = data
         self.dircache.clear()  # this is a bit heavy handed
 
-    async def _put_file(self, lpath, rpath, **kwargs):
+    async def _put_file(self, lpath, rpath, mode="overwrite", **kwargs):
         # puts binary
+        if mode == "create" and self.exists(rpath):
+            raise FileExistsError
         with open(lpath, "rb") as f:
             self.references[rpath] = f.read()
         self.dircache.clear()  # this is a bit heavy handed
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/spec.py b/fsspec/spec.py
index 9659f2e..6837ab3 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -10,7 +10,7 @@ import weakref
 from errno import ESPIPE
 from glob import has_magic
 from hashlib import sha256
-from typing import Any, ClassVar, Dict, Tuple
+from typing import Any, ClassVar
 
 from .callbacks import DEFAULT_CALLBACK
 from .config import apply_config, conf
@@ -117,8 +117,8 @@ class AbstractFileSystem(metaclass=_Cached):
     _extra_tokenize_attributes = ()
 
     # Set by _Cached metaclass
-    storage_args: Tuple[Any, ...]
-    storage_options: Dict[str, Any]
+    storage_args: tuple[Any, ...]
+    storage_options: dict[str, Any]
 
     def __init__(self, *args, **storage_options):
         """Create and configure file-system instance
@@ -615,11 +615,9 @@ class AbstractFileSystem(metaclass=_Cached):
             p: info
             for p, info in sorted(allpaths.items())
             if pattern.match(
-                (
-                    p + "/"
-                    if append_slash_to_dirname and info["type"] == "directory"
-                    else p
-                )
+                p + "/"
+                if append_slash_to_dirname and info["type"] == "directory"
+                else p
             )
         }
 
@@ -780,8 +778,12 @@ class AbstractFileSystem(metaclass=_Cached):
                 return f.read(end - f.tell())
             return f.read()
 
-    def pipe_file(self, path, value, **kwargs):
+    def pipe_file(self, path, value, mode="overwrite", **kwargs):
         """Set the bytes of given file"""
+        if mode == "create" and self.exists(path):
+            # non-atomic but simple way; or could use "xb" in open(), which is likely
+            # not as well supported
+            raise FileExistsError
         with self.open(path, "wb", **kwargs) as f:
             f.write(value)
 
@@ -973,8 +975,12 @@ class AbstractFileSystem(metaclass=_Cached):
             with callback.branched(rpath, lpath) as child:
                 self.get_file(rpath, lpath, callback=child, **kwargs)
 
-    def put_file(self, lpath, rpath, callback=DEFAULT_CALLBACK, **kwargs):
+    def put_file(
+        self, lpath, rpath, callback=DEFAULT_CALLBACK, mode="overwrite", **kwargs
+    ):
         """Copy single file to remote"""
+        if mode == "create" and self.exists(rpath):
+            raise FileExistsError
         if os.path.isdir(lpath):
             self.makedirs(rpath, exist_ok=True)
             return None
@@ -1264,6 +1270,9 @@ class AbstractFileSystem(metaclass=_Cached):
             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
@@ -1442,7 +1451,7 @@ class AbstractFileSystem(metaclass=_Cached):
 
         return json.loads(blob, cls=FilesystemJSONDecoder)
 
-    def to_dict(self, *, include_password: bool = True) -> Dict[str, Any]:
+    def to_dict(self, *, include_password: bool = True) -> dict[str, Any]:
         """
         JSON-serializable dictionary representation of this filesystem instance.
 
@@ -1483,7 +1492,7 @@ class AbstractFileSystem(metaclass=_Cached):
         )
 
     @staticmethod
-    def from_dict(dct: Dict[str, Any]) -> AbstractFileSystem:
+    def from_dict(dct: dict[str, Any]) -> AbstractFileSystem:
         """
         Recreate a filesystem instance from dictionary representation.
 
@@ -1797,7 +1806,7 @@ class AbstractBufferedFile(io.IOBase):
 
     def info(self):
         """File information about this path"""
-        if "r" in self.mode:
+        if self.readable():
             return self.details
         else:
             raise ValueError("Info not available while writing")
@@ -1844,7 +1853,7 @@ class AbstractBufferedFile(io.IOBase):
         data: bytes
             Set of bytes to be written.
         """
-        if self.mode not in {"wb", "ab"}:
+        if not self.writable():
             raise ValueError("File not in write mode")
         if self.closed:
             raise ValueError("I/O operation on closed file.")
@@ -1877,7 +1886,7 @@ class AbstractBufferedFile(io.IOBase):
         if force:
             self.forced = True
 
-        if self.mode not in {"wb", "ab"}:
+        if self.readable():
             # no-op to flush on read-mode
             return
 
@@ -2026,21 +2035,22 @@ class AbstractBufferedFile(io.IOBase):
             return
         if self.closed:
             return
-        if self.mode == "rb":
-            self.cache = None
-        else:
-            if not self.forced:
-                self.flush(force=True)
-
-            if self.fs is not None:
-                self.fs.invalidate_cache(self.path)
-                self.fs.invalidate_cache(self.fs._parent(self.path))
+        try:
+            if self.mode == "rb":
+                self.cache = None
+            else:
+                if not self.forced:
+                    self.flush(force=True)
 
-        self.closed = True
+                if self.fs is not None:
+                    self.fs.invalidate_cache(self.path)
+                    self.fs.invalidate_cache(self.fs._parent(self.path))
+        finally:
+            self.closed = True
 
     def readable(self):
         """Whether opened for reading"""
-        return self.mode == "rb" and not self.closed
+        return "r" in self.mode and not self.closed
 
     def seekable(self):
         """Whether is seekable (only in read mode)"""
@@ -2048,7 +2058,7 @@ class AbstractBufferedFile(io.IOBase):
 
     def writable(self):
         """Whether opened for writing"""
-        return self.mode in {"wb", "ab"} and not self.closed
+        return self.mode in {"wb", "ab", "xb"} and not self.closed
 
     def __del__(self):
         if not self.closed:
diff --git a/fsspec/utils.py b/fsspec/utils.py
index faa6393..917af8d 100644
--- a/fsspec/utils.py
+++ b/fsspec/utils.py
@@ -4,7 +4,6 @@ import contextlib
 import logging
 import math
 import os
-import pathlib
 import re
 import sys
 import tempfile
@@ -24,6 +23,8 @@ from typing import (
 from urllib.parse import urlsplit
 
 if TYPE_CHECKING:
+    import pathlib
+
     from typing_extensions import TypeGuard
 
     from fsspec.spec import AbstractFileSystem
@@ -82,7 +83,8 @@ def infer_storage_options(
         # https://msdn.microsoft.com/en-us/library/jj710207.aspx
         windows_path = re.match(r"^/([a-zA-Z])[:|]([\\/].*)$", path)
         if windows_path:
-            path = "%s:%s" % windows_path.groups()
+            drive, path = windows_path.groups()
+            path = f"{drive}:{path}"
 
     if protocol in ["http", "https"]:
         # for HTTP, we don't want to parse, as requests will anyway
diff --git a/pyproject.toml b/pyproject.toml
index d9cb2d6..7ec3c85 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -163,6 +163,7 @@ select = [
     "E9",
     "F",
     "FURB",
+    "I",
     "LOG",
     "PERF",
     "PLC",
@@ -181,8 +182,11 @@ select = [
     "RUF006",
     "RUF015",
     "RUF024",
+    "SIM",
     "SLOT",
     "SIM101",
+    "TCH",
+    "UP",
 ]
 ignore = [
 	# Loop control variable `loop` not used within loop body
@@ -207,6 +211,18 @@ ignore = [
     # Fix these codes later
     "G004",
     "PERF203",
+    "UP007",
+    "UP011",
+    "UP015",
+    "UP018",
+    # deprecated
+    "UP027",
+    "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/memory/memory_test.py b/fsspec/implementations/tests/memory/memory_test.py
index fd0ebaa..c165d4d 100644
--- a/fsspec/implementations/tests/memory/memory_test.py
+++ b/fsspec/implementations/tests/memory/memory_test.py
@@ -12,3 +12,11 @@ class TestMemoryGet(abstract.AbstractGetTests, MemoryFixtures):
 
 class TestMemoryPut(abstract.AbstractPutTests, MemoryFixtures):
     pass
+
+
+class TestMemoryPipe(abstract.AbstractPipeTests, MemoryFixtures):
+    pass
+
+
+class TestMemoryOpen(abstract.AbstractOpenTests, MemoryFixtures):
+    pass
diff --git a/fsspec/implementations/tests/test_asyn_wrapper.py b/fsspec/implementations/tests/test_asyn_wrapper.py
index d292020..82e9f3d 100644
--- a/fsspec/implementations/tests/test_asyn_wrapper.py
+++ b/fsspec/implementations/tests/test_asyn_wrapper.py
@@ -1,10 +1,12 @@
 import asyncio
-import pytest
 import os
 
+import pytest
+
 import fsspec
 from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
 from fsspec.implementations.local import LocalFileSystem
+
 from .test_local import csv_files, filetexts
 
 
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_smb.py b/fsspec/implementations/tests/test_smb.py
index a83e3cc..a4b4435 100644
--- a/fsspec/implementations/tests/test_smb.py
+++ b/fsspec/implementations/tests/test_smb.py
@@ -14,6 +14,12 @@ import fsspec
 
 pytest.importorskip("smbprotocol")
 
+
+def delay_rerun(*args):
+    time.sleep(0.1)
+    return True
+
+
 # ruff: noqa: F821
 
 if os.environ.get("WSL_INTEROP"):
@@ -29,7 +35,7 @@ else:
 
 
 def stop_docker(container):
-    cmd = shlex.split('docker ps -a -q --filter "name=%s"' % container)
+    cmd = shlex.split(f'docker ps -a -q --filter "name={container}"')
     cid = subprocess.check_output(cmd).strip().decode()
     if cid:
         subprocess.call(["docker", "rm", "-f", "-v", cid])
@@ -72,7 +78,7 @@ def smb_params(request):
         stop_docker(container)
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_simple(smb_params):
     adir = "/home/adir"
     adir2 = "/home/adir/otherdir/"
@@ -89,7 +95,7 @@ def test_simple(smb_params):
     assert not fsmb.exists(adir)
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_auto_mkdir(smb_params):
     adir = "/home/adir"
     adir2 = "/home/adir/otherdir/"
@@ -116,7 +122,7 @@ def test_auto_mkdir(smb_params):
     assert not fsmb.exists(another_dir)
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_with_url(smb_params):
     if smb_params["port"] is None:
         smb_url = "smb://{username}:{password}@{host}/home/someuser.txt"
@@ -131,7 +137,7 @@ def test_with_url(smb_params):
         assert read_result == b"hello"
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_transaction(smb_params):
     afile = "/home/afolder/otherdir/afile"
     afile2 = "/home/afolder/otherdir/afile2"
@@ -152,14 +158,14 @@ def test_transaction(smb_params):
     assert fsmb.find(adir) == [afile, afile2]
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_makedirs_exist_ok(smb_params):
     fsmb = fsspec.get_filesystem_class("smb")(**smb_params)
     fsmb.makedirs("/home/a/b/c")
     fsmb.makedirs("/home/a/b/c", exist_ok=True)
 
 
-@pytest.mark.flaky(reruns=2, reruns_delay=2)
+@pytest.mark.flaky(max_runs=3, rerun_filter=delay_rerun)
 def test_rename_from_upath(smb_params):
     fsmb = fsspec.get_filesystem_class("smb")(**smb_params)
     fsmb.makedirs("/home/a/b/c", exist_ok=True)
diff --git a/fsspec/implementations/tests/test_tar.py b/fsspec/implementations/tests/test_tar.py
index 0ec7c8a..e74f54d 100644
--- a/fsspec/implementations/tests/test_tar.py
+++ b/fsspec/implementations/tests/test_tar.py
@@ -5,7 +5,7 @@ import shutil
 import tarfile
 import tempfile
 from io import BytesIO
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 import pytest
 
@@ -15,6 +15,9 @@ from fsspec.implementations.cached import WholeFileCacheFileSystem
 from fsspec.implementations.tar import TarFileSystem
 from fsspec.implementations.tests.test_archive import archive_data, temptar
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 def test_info():
     with temptar(archive_data) as t:
diff --git a/fsspec/implementations/tests/test_zip.py b/fsspec/implementations/tests/test_zip.py
index 14d0008..59559c0 100644
--- a/fsspec/implementations/tests/test_zip.py
+++ b/fsspec/implementations/tests/test_zip.py
@@ -105,7 +105,7 @@ def test_zip_glob_star(m):
     outfiles = fs.glob("*")
     assert len(outfiles) == 1
 
-    fn = f"{os.path.dirname(os.path.abspath((__file__)))}/out.zip"
+    fn = f"{os.path.dirname(os.path.abspath(__file__))}/out.zip"
     fs = fsspec.filesystem("zip", fo=fn, mode="r")
     outfiles = fs.glob("*")
     assert len(outfiles) == 1
@@ -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/abstract/__init__.py b/fsspec/tests/abstract/__init__.py
index 4418142..8ed2ad8 100644
--- a/fsspec/tests/abstract/__init__.py
+++ b/fsspec/tests/abstract/__init__.py
@@ -6,6 +6,8 @@ import pytest
 from fsspec.implementations.local import LocalFileSystem
 from fsspec.tests.abstract.copy import AbstractCopyTests  # noqa: F401
 from fsspec.tests.abstract.get import AbstractGetTests  # noqa: F401
+from fsspec.tests.abstract.open import AbstractOpenTests  # noqa: F401
+from fsspec.tests.abstract.pipe import AbstractPipeTests  # noqa: F401
 from fsspec.tests.abstract.put import AbstractPutTests  # noqa: F401
 
 
@@ -225,7 +227,7 @@ class BaseAbstractFixtures:
         for i in range(10):
             hashed_i = md5(str(i).encode("utf-8")).hexdigest()
             path = some_join(source, f"{hashed_i}.txt")
-            some_fs.pipe(path=path, value=f"{i}".encode("utf-8"))
+            some_fs.pipe(path=path, value=f"{i}".encode())
         return source
 
 
diff --git a/fsspec/tests/abstract/open.py b/fsspec/tests/abstract/open.py
new file mode 100644
index 0000000..bb75ea8
--- /dev/null
+++ b/fsspec/tests/abstract/open.py
@@ -0,0 +1,11 @@
+import pytest
+
+
+class AbstractOpenTests:
+    def test_open_exclusive(self, fs, fs_target):
+        with fs.open(fs_target, "wb") as f:
+            f.write(b"data")
+        with fs.open(fs_target, "rb") as f:
+            assert f.read() == b"data"
+        with pytest.raises(FileExistsError):
+            fs.open(fs_target, "xb")
diff --git a/fsspec/tests/abstract/pipe.py b/fsspec/tests/abstract/pipe.py
new file mode 100644
index 0000000..8ecca96
--- /dev/null
+++ b/fsspec/tests/abstract/pipe.py
@@ -0,0 +1,11 @@
+import pytest
+
+
+class AbstractPipeTests:
+    def test_pipe_exclusive(self, fs, fs_target):
+        fs.pipe_file(fs_target, b"data")
+        assert fs.cat_file(fs_target) == b"data"
+        with pytest.raises(FileExistsError):
+            fs.pipe_file(fs_target, b"data", mode="create")
+        fs.pipe_file(fs_target, b"new data", mode="overwrite")
+        assert fs.cat_file(fs_target) == b"new data"
diff --git a/fsspec/tests/test_async.py b/fsspec/tests/test_async.py
index aa3c9bd..4282996 100644
--- a/fsspec/tests/test_async.py
+++ b/fsspec/tests/test_async.py
@@ -208,7 +208,7 @@ class DummyAsyncStreamedFile(fsspec.asyn.AbstractAsyncStreamedFile):
 async def test_async_streamed_file_write():
     test_fs = DummyAsyncFS()
     streamed_file = await test_fs.open_async("misc/foo.txt", mode="wb")
-    inp_data = "foo-bar".encode("utf8") * streamed_file.blocksize * 2
+    inp_data = b"foo-bar" * streamed_file.blocksize * 2
     await streamed_file.write(inp_data)
     assert streamed_file.loc == len(inp_data)
     await streamed_file.close()
diff --git a/fsspec/tests/test_fuse.py b/fsspec/tests/test_fuse.py
index 5222d2f..ef30053 100644
--- a/fsspec/tests/test_fuse.py
+++ b/fsspec/tests/test_fuse.py
@@ -121,8 +121,7 @@ def test_chmod(mount_local):
 
     cp = subprocess.run(
         ["cp", str(mount_dir / "text"), str(mount_dir / "new")],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
+        capture_output=True,
         check=False,
     )
 
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)
diff --git a/fsspec/tests/test_utils.py b/fsspec/tests/test_utils.py
index 1eeee91..c105546 100644
--- a/fsspec/tests/test_utils.py
+++ b/fsspec/tests/test_utils.py
@@ -45,9 +45,7 @@ def test_read_block():
 
 def test_read_block_split_before():
     """Test start/middle/end cases of split_before."""
-    d = (
-        "#header" + "".join(">foo{i}\nFOOBAR{i}\n".format(i=i) for i in range(100000))
-    ).encode()
+    d = ("#header" + "".join(f">foo{i}\nFOOBAR{i}\n" for i in range(100000))).encode()
 
     # Read single record at beginning.
     # All reads include beginning of file and read through termination of