Repository: fsspec/filesystem_spec
Pull request #1750 (2024-11-14 14:32:13)

Add tree method to display tree-like structure of the filesystem
>>> fs.tree(path='/start/folder', display_size=True)

/start/folder
├── folder1
│   ├── file1.txt (1.234MB)
│   └── file2.txt (0.567MB)
└── folder2
    └── file3.txt (2.345MB)

CyrilJl: @martindurant I went along with your proposals. This can be used as follows:
```python
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)

/
├────── pub (1 subfolder)
│       └────── example (16 files)
│               ├────── KeyGenerator.png (35.81 kb)
│               ├────── KeyGeneratorSmall.png (23.47 kb)
│               ├────── ResumableTransfer.png (11.28 kb)
│               ├────── WinFormClient.png (78.12 kb)
│               ├────── WinFormClientSmall.png (17.49 kb)
│               ├────── imap-console-client.png (18.71 kb)
│               ├────── mail-editor.png (16.08 kb)
│               ├────── mail-send-winforms.png (34.58 kb)
│               ├────── mime-explorer.png (47.86 kb)
│               ├────── pocketftp.png (56.66 kb)
│               └────── 6 more item(s) not displayed.
└────── readme.txt (379B)
```

Or:

```python
fs = filesystem('gs')
tree = fs.tree(path='gs://gcp-public-data-arco-era5', display_size=True, recursion_limit=2, indent_size=8, max_display=5)
print(tree)

gs://gcp-public-data-arco-era5
├────── ar (19 subfolders)
│       ├────── 1959-2022-1h-240x121_equiangular_with_poles_conservative.zarr (3 files, 35 subfolders)
│       ├────── 1959-2022-1h-360x181_equiangular_with_poles_conservative.zarr (3 files, 35 subfolders)
│       ├────── 1959-2022-6h-128x64_equiangular_conservative.zarr (3 files, 42 subfolders)
│       ├────── 1959-2022-6h-128x64_equiangular_with_poles_conservative.zarr (3 files, 42 subfolders)
│       ├────── 1959-2022-6h-1440x721.zarr (3 files, 42 subfolders)
│       └────── 14 more item(s) not displayed.
├────── co (10 subfolders)
│       ├────── model-level-moisture.zarr (3 files, 13 subfolders)
│       ├────── model-level-moisture.zarr-v2 (3 files, 13 subfolders)
│       ├────── model-level-wind.zarr (3 files, 8 subfolders)
│       ├────── model-level-wind.zarr-v2 (3 files, 8 subfolders)
│       ├────── single-level-forecast.zarr (3 files, 28 subfolders)
│       └────── 5 more item(s) not displayed.
└────── raw (4 subfolders)
        ├────── ERA5GRIB (1 subfolder)
        ├────── date-variable-pressure_level (85 subfolders)
        ├────── date-variable-single_level (85 subfolders)
        └────── date-variable-static (1 subfolder)
```

Is this what you had in mind ?
martindurant: Looks good! 

martindurant: It would still be nice to create some minimal test for this, based on files in memoryFS 
martindurant: (please run `pre-commit run -a` here; I can't do this because "master" is a taken branch name)
martindurant: See https://github.com/fsspec/filesystem_spec/pull/1755 for mention of the s3fs failure

Patch:
diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index f9e5ed9..23dc746 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -156,6 +156,7 @@ jobs:
         run: |
           pip install -e . --no-deps
           pip install -e ./${{ matrix.FRIEND }} --no-deps
+          pip list
 
       - name: Test
         shell: bash -l {0}
diff --git a/ci/environment-friends.yml b/ci/environment-friends.yml
index 114c854..6e6b79a 100644
--- a/ci/environment-friends.yml
+++ b/ci/environment-friends.yml
@@ -9,7 +9,6 @@ dependencies:
   - pytest-cov
   - pytest-mock
   - pip
-  - pytest<8
   - ujson
   - requests
   - decorator
@@ -25,5 +24,6 @@ dependencies:
   - google-api-python-client
   - httpretty
   - aiobotocore
-  - "moto >=4,<5"
+  - moto
   - flask
+  - pyopenssl>=24.0.0
diff --git a/docs/source/features.rst b/docs/source/features.rst
index 907084e..949e471 100644
--- a/docs/source/features.rst
+++ b/docs/source/features.rst
@@ -408,3 +408,21 @@ tqdm.
 
     <script data-goatcounter="https://fsspec.goatcounter.com/count"
         async src="//gc.zgo.at/count.js"></script>
+
+
+Exclusive write
+---------------
+
+Some backends support writing to a file only if it doesn't already exist. This may be
+implemented for the following methods:
+- pipe_file (with argument ``mode=='create'``)
+- put_file (with argument ``mode=='create'``)
+- open (with argument ``mode="xb"``)
+Since some writes will be achieved in blocks, the timing of when the check is done is
+not defined - it may be at the start or at the completion of the operation, depending
+on the backend.
+
+If using exclusive mode on a file that does already exist, a ``FileExistsError`` will
+be raised.
+
+This feature is currently included on a trial basis and may change in the future.
diff --git a/fsspec/implementations/asyn_wrapper.py b/fsspec/implementations/asyn_wrapper.py
index 0d46265..43604a2 100644
--- a/fsspec/implementations/asyn_wrapper.py
+++ b/fsspec/implementations/asyn_wrapper.py
@@ -45,26 +45,27 @@ class AsyncFileSystemWrapper(AsyncFileSystem):
     def __init__(self, sync_fs, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.asynchronous = True
-        self.fs = sync_fs
+        self.sync_fs = sync_fs
+        self.protocol = self.sync_fs.protocol
         self._wrap_all_sync_methods()
 
     @property
     def fsid(self):
-        return f"async_{self.fs.fsid}"
+        return f"async_{self.sync_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):
+        for method_name in dir(self.sync_fs):
             if method_name.startswith("_"):
                 continue
 
-            attr = inspect.getattr_static(self.fs, method_name)
+            attr = inspect.getattr_static(self.sync_fs, method_name)
             if isinstance(attr, property):
                 continue
 
-            method = getattr(self.fs, method_name)
+            method = getattr(self.sync_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)
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index 5c5372c..60fb5d5 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -20,6 +20,7 @@ except ImportError:
 from fsspec.asyn import AsyncFileSystem
 from fsspec.callbacks import DEFAULT_CALLBACK
 from fsspec.core import filesystem, open, split_protocol
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
 from fsspec.utils import isfilelike, merge_offset_ranges, other_paths
 
 logger = logging.getLogger("fsspec.reference")
@@ -757,6 +758,10 @@ class ReferenceFileSystem(AsyncFileSystem):
             self.fss[remote_protocol] = fs
 
         self.fss[None] = fs or filesystem("file")  # default one
+        # Wrap any non-async filesystems to ensure async methods are available below
+        for k, f in self.fss.items():
+            if not f.async_impl:
+                self.fss[k] = AsyncFileSystemWrapper(f)
 
     def _cat_common(self, path, start=None, end=None):
         path = self._strip_protocol(path)
diff --git a/fsspec/implementations/webhdfs.py b/fsspec/implementations/webhdfs.py
index 300bb9c..c6e0d84 100644
--- a/fsspec/implementations/webhdfs.py
+++ b/fsspec/implementations/webhdfs.py
@@ -166,7 +166,8 @@ class WebHDFS(AbstractFileSystem):
             self.session.auth = HTTPBasicAuth(self.user, self.password)
 
     def _call(self, op, method="get", path=None, data=None, redirect=True, **kwargs):
-        url = self._apply_proxy(self.url + quote(path or "", safe="/="))
+        path = self._strip_protocol(path) if path is not None else ""
+        url = self._apply_proxy(self.url + quote(path, safe="/="))
         args = kwargs.copy()
         args.update(self.pars)
         args["op"] = op.upper()
diff --git a/fsspec/spec.py b/fsspec/spec.py
index ad8703b..7daf850 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -1576,6 +1576,141 @@ class AbstractFileSystem(metaclass=_Cached):
         """Return the modified timestamp of a file as a datetime.datetime"""
         raise NotImplementedError
 
+    def tree(
+        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)
+        """
+
+        def format_bytes(n: int) -> str:
+            """Format bytes as text."""
+            for prefix, k in (
+                ("P", 2**50),
+                ("T", 2**40),
+                ("G", 2**30),
+                ("M", 2**20),
+                ("k", 2**10),
+            ):
+                if n >= 0.9 * k:
+                    return f"{n / k:.2f} {prefix}b"
+            return f"{n}B"
+
+        result = []
+
+        if first:
+            result.append(path)
+
+        if recursion_limit:
+            indent = " " * indent_size
+            contents = self.ls(path, detail=True)
+            contents.sort(
+                key=lambda x: (x.get("type") != "directory", x.get("name", ""))
+            )
+
+            if max_display is not None and len(contents) > max_display:
+                displayed_contents = contents[:max_display]
+                remaining_count = len(contents) - max_display
+            else:
+                displayed_contents = contents
+                remaining_count = 0
+
+            for i, item in enumerate(displayed_contents):
+                is_last_item = (i == len(displayed_contents) - 1) and (
+                    remaining_count == 0
+                )
+
+                branch = (
+                    "└" + ("─" * (indent_size - 2))
+                    if is_last_item
+                    else "├" + ("─" * (indent_size - 2))
+                )
+                branch += " "
+                new_prefix = prefix + (
+                    indent if is_last_item else "│" + " " * (indent_size - 1)
+                )
+
+                name = os.path.basename(item.get("name", ""))
+
+                if display_size and item.get("type") == "directory":
+                    sub_contents = self.ls(item.get("name", ""), detail=True)
+                    num_files = sum(
+                        1 for sub_item in sub_contents if sub_item.get("type") == "file"
+                    )
+                    num_folders = sum(
+                        1
+                        for sub_item in sub_contents
+                        if sub_item.get("type") == "directory"
+                    )
+
+                    if num_files == 0 and num_folders == 0:
+                        size = " (empty folder)"
+                    elif num_files == 0:
+                        size = f" ({num_folders} subfolder{'s' if num_folders > 1 else ''})"
+                    elif num_folders == 0:
+                        size = f" ({num_files} file{'s' if num_files > 1 else ''})"
+                    else:
+                        size = f" ({num_files} file{'s' if num_files > 1 else ''}, {num_folders} subfolder{'s' if num_folders > 1 else ''})"
+                elif display_size and item.get("type") == "file":
+                    size = f" ({format_bytes(item.get('size', 0))})"
+                else:
+                    size = ""
+
+                result.append(f"{prefix}{branch}{name}{size}")
+
+                if item.get("type") == "directory" and recursion_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,
+                        )
+                    )
+
+            if remaining_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 _ in result if _)
+
     # ------------------------------------------------------------------------
     # Aliases
 


Test patch:
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index 0063d11..dd9ff68 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -458,7 +458,8 @@ def test_fss_has_defaults(m):
     assert fs.fss["memory"].protocol == "memory"
 
     fs = fsspec.filesystem("reference", fs=m, fo={})
-    assert fs.fss[None] is m
+    # Default behavior here wraps synchronous filesystems to enable the async API
+    assert fs.fss[None].sync_fs is m
 
     fs = fsspec.filesystem("reference", fs={"memory": m}, fo={})
     assert fs.fss["memory"] is m
diff --git a/fsspec/implementations/tests/test_webhdfs.py b/fsspec/implementations/tests/test_webhdfs.py
index fac34c7..0f179d3 100644
--- a/fsspec/implementations/tests/test_webhdfs.py
+++ b/fsspec/implementations/tests/test_webhdfs.py
@@ -195,3 +195,16 @@ def test_write_read_verify_file_with_equals(hdfs_cluster):
     assert len(file_info) == 1
     assert file_info[0]["name"] == file_path
     assert file_info[0]["size"] == len(content)
+
+
+def test_protocol_prefixed_path(hdfs_cluster):
+    fs = WebHDFS(
+        hdfs_cluster, user="testuser", data_proxy={"worker.example.com": "localhost"}
+    )
+    protocol_prefixed_path = "webhdfs://localhost:50070/user/testuser/test_dir"
+
+    fs.mkdir(protocol_prefixed_path)
+    assert fs.exists(protocol_prefixed_path)
+
+    file_info = fs.ls(protocol_prefixed_path, detail=True)
+    assert len(file_info) == 0