Mark dir, async wrapper fs as chained

fsspec/core.py

@@ -18,7 +18,7 @@
 )
 from fsspec.compression import compr
 from fsspec.config import conf
-from fsspec.registry import filesystem, get_filesystem_class
+from fsspec.registry import available_protocols, filesystem, get_filesystem_class
 from fsspec.utils import (
     _unstrip_protocol,
     build_name_function,
@@ -334,34 +334,51 @@ def _un_chain(path, kwargs):
 
     if "::" in path:
         x = re.compile(".*[^a-z]+.*")  # test for non protocol-like single word
+        known_protocols = set(available_protocols())
         bits = []
+
+        # split on '::', then ensure each bit has a protocol
         for p in path.split("::"):
-            if "://" in p or x.match(p):
+            if p in known_protocols:
+                bits.append(p + "://")
+            elif "://" in p or x.match(p):
                 bits.append(p)
             else:
                 bits.append(p + "://")
     else:
         bits = [path]
+
     # [[url, protocol, kwargs], ...]
     out = []
     previous_bit = None
     kwargs = kwargs.copy()
+
     for bit in reversed(bits):
         protocol = kwargs.pop("protocol", None) or split_protocol(bit)[0] or "file"
         cls = get_filesystem_class(protocol)
         extra_kwargs = cls._get_kwargs_from_urls(bit)
         kws = kwargs.pop(protocol, {})
+
         if bit is bits[0]:
             kws.update(kwargs)
+
         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 "target_protocol" not in kw and issubclass(cls, ChainedFileSystem):
+
+        if (
+            "target_protocol" not in kw
+            and issubclass(cls, ChainedFileSystem)
+            and not bit
+        ):
+            # replace bit if we are chaining and no path given
             bit = previous_bit
+
         out.append((bit, protocol, kw))
         previous_bit = bit
+
     out.reverse()
     return out
 

fsspec/implementations/asyn_wrapper.py

@@ -5,6 +5,8 @@
 import fsspec
 from fsspec.asyn import AsyncFileSystem, running_async
 
+from .chained import ChainedFileSystem
+
 
 def async_wrapper(func, obj=None, semaphore=None):
     """
@@ -35,7 +37,7 @@ async def wrapper(*args, **kwargs):
     return wrapper
 
 
-class AsyncFileSystemWrapper(AsyncFileSystem):
+class AsyncFileSystemWrapper(AsyncFileSystem, ChainedFileSystem):
     """
     A wrapper class to convert a synchronous filesystem into an asynchronous one.
 

fsspec/implementations/dirfs.py

@@ -1,8 +1,9 @@
 from .. import filesystem
 from ..asyn import AsyncFileSystem
+from .chained import ChainedFileSystem
 
 
-class DirFileSystem(AsyncFileSystem):
+class DirFileSystem(AsyncFileSystem, ChainedFileSystem):
     """Directory prefix filesystem
 
     The DirFileSystem is a filesystem-wrapper. It assumes every path it is dealing with

fsspec/registry.py

@@ -72,6 +72,9 @@ def register_implementation(name, cls, clobber=False, errtxt=None):
         "class": "fsspec.implementations.arrow.HadoopFileSystem",
         "err": "pyarrow and local java libraries required for HDFS",
     },
+    "async_wrapper": {
+        "class": "fsspec.implementations.asyn_wrapper.AsyncFileSystemWrapper",
+    },
     "asynclocal": {
         "class": "morefs.asyn_local.AsyncLocalFileSystem",
         "err": "Install 'morefs[asynclocalfs]' to use AsyncLocalFileSystem",