Implement UPath.resolve (#86)

* Stop parsing URI paths as posix paths

* Add URI path normalization in `UPath.resolve`

* Follow redirects in `HTTPPath.resolve`

* Update test_http.py

* linting

* trigger ci

---------

Co-authored-by: Norman Rzepka <[email protected]>

Author: joouha

upath/core.py

@@ -4,6 +4,7 @@
 import re
 import sys
 from os import PathLike
+from pathlib import _PosixFlavour  # type: ignore
 from typing import Sequence
 from typing import TypeVar
 from typing import TYPE_CHECKING
@@ -89,17 +90,41 @@ def touch(self, path, **kwargs):
         return self._fs.touch(self._format_path(path), **kwargs)
 
 
+class _UriFlavour(_PosixFlavour):
+    def parse_parts(self, parts):
+        parsed = []
+        sep = self.sep
+        drv = root = ""
+        it = reversed(parts)
+        for part in it:
+            if part:
+                drv, root, rel = self.splitroot(part)
+                if not root or root and rel:
+                    for x in reversed(rel.split(sep)):
+                        parsed.append(sys.intern(x))
+
+        if drv or root:
+            parsed.append(drv + root)
+        parsed.reverse()
+        return drv, root, parsed
+
+    def splitroot(self, part, sep="/"):
+        # Treat the first slash in the path as the root if it exists
+        if part and part[0] == sep:
+            return "", sep, part[1:]
+        return "", "", part
+
+
 PT = TypeVar("PT", bound="UPath")
 
 
 class UPath(pathlib.Path):
-
     __slots__ = (
         "_url",
         "_kwargs",
         "_accessor",  # overwritten because of default in Python 3.10
     )
-    _flavour = pathlib._posix_flavour  # type: ignore
+    _flavour = _UriFlavour()
     _default_accessor = _FSSpecAccessor
 
     # typing
@@ -311,7 +336,40 @@ def absolute(self: PT) -> PT:
         return self
 
     def resolve(self: PT, strict: bool = False) -> PT:
-        raise NotImplementedError
+        """Return a new path with '.' and '..' parts normalized."""
+        _parts = self._parts
+
+        # Do not attempt to normalize path if no parts are dots
+        if ".." not in _parts and "." not in _parts:
+            return self
+
+        sep = self._flavour.sep
+
+        resolved: list[str] = []
+        resolvable_parts = _parts[1:]
+        idx_max = len(resolvable_parts) - 1
+        for i, part in enumerate(resolvable_parts):
+            if part == "..":
+                if resolved:
+                    resolved.pop()
+            elif part != ".":
+                if i < idx_max:
+                    part += sep
+                resolved.append(part)
+
+        path = "".join(resolved)
+        url = self._url
+        if url is not None:
+            url = url._replace(path=path)
+        parts = _parts[:1] + path.split(sep)
+
+        return self._from_parsed_parts(
+            self._drv,
+            self._root,
+            parts,
+            url=url,
+            **self._kwargs,
+        )
 
     def exists(self) -> bool:
         """

upath/implementations/http.py

@@ -1,3 +1,5 @@
+from fsspec.asyn import sync
+
 import upath.core
 
 
@@ -47,3 +49,35 @@ def _sub_path(self, name):
         name = name.strip("/")
 
         return name
+
+    def resolve(
+        self: "HTTPPath", strict: bool = False, follow_redirects: bool = True
+    ) -> "HTTPPath":
+        """Normalize the path and resolve redirects."""
+        # Normalise the path
+        resolved_path = super().resolve(strict=strict)
+
+        if follow_redirects:
+            # Ensure we have a url
+            parsed_url = resolved_path._url
+            if parsed_url is None:
+                return resolved_path
+            else:
+                url = parsed_url.geturl()
+            # Get the fsspec fs
+            fs = resolved_path._accessor._fs
+            # Ensure we have a session
+            session = sync(fs.loop, fs.set_session)
+            # Use HEAD requests if the server allows it, falling back to GETs
+            for method in (session.head, session.get):
+                r = sync(fs.loop, method, url, allow_redirects=True)
+                try:
+                    r.raise_for_status()
+                except Exception as exc:
+                    if method == session.get:
+                        raise FileNotFoundError(self) from exc
+                else:
+                    resolved_path = HTTPPath(str(r.url))
+                    break
+
+        return resolved_path

upath/tests/implementations/test_http.py

@@ -76,3 +76,9 @@ def test_write_text(self, pathlib_base):
 
     def test_fsspec_compat(self):
         pass
+
+    def test_resolve(self):
+        # Also tests following redirects, because the test server issues a
+        # 301 redirect for `http://127.0.0.1:8080/folder` to
+        # `http://127.0.0.1:8080/folder/`
+        assert str(self.path.resolve()).endswith("/")

upath/tests/test_core.py

@@ -251,3 +251,71 @@ def test_relative_to():
         UPath("s3://test_bucket/file.txt", anon=True).relative_to(
             UPath("s3://test_bucket", anon=False)
         )
+
+
+def test_uri_parsing():
+    assert (
+        str(UPath("http://www.example.com//a//b/"))
+        == "http://www.example.com//a//b/"
+    )
+
+
+NORMALIZATIONS = (
+    ("unnormalized", "normalized"),
+    (
+        # Expected normalization results according to curl
+        ("http://example.com", "http://example.com/"),
+        ("http://example.com/", "http://example.com/"),
+        ("http://example.com/a", "http://example.com/a"),
+        ("http://example.com//a", "http://example.com//a"),
+        ("http://example.com///a", "http://example.com///a"),
+        ("http://example.com////a", "http://example.com////a"),
+        ("http://example.com/a/.", "http://example.com/a/"),
+        ("http://example.com/a/./", "http://example.com/a/"),
+        ("http://example.com/a/./b", "http://example.com/a/b"),
+        ("http://example.com/a/.//", "http://example.com/a//"),
+        ("http://example.com/a/.//b", "http://example.com/a//b"),
+        ("http://example.com/a//.", "http://example.com/a//"),
+        ("http://example.com/a//./", "http://example.com/a//"),
+        ("http://example.com/a//./b", "http://example.com/a//b"),
+        ("http://example.com/a//.//", "http://example.com/a///"),
+        ("http://example.com/a//.//b", "http://example.com/a///b"),
+        ("http://example.com/a/..", "http://example.com/"),
+        ("http://example.com/a/../", "http://example.com/"),
+        ("http://example.com/a/../.", "http://example.com/"),
+        ("http://example.com/a/../..", "http://example.com/"),
+        ("http://example.com/a/../../", "http://example.com/"),
+        ("http://example.com/a/../..//", "http://example.com//"),
+        ("http://example.com/a/..//", "http://example.com//"),
+        ("http://example.com/a/..//.", "http://example.com//"),
+        ("http://example.com/a/..//..", "http://example.com/"),
+        ("http://example.com/a/../b", "http://example.com/b"),
+        ("http://example.com/a/..//b", "http://example.com//b"),
+        ("http://example.com/a//..", "http://example.com/a/"),
+        ("http://example.com/a//../", "http://example.com/a/"),
+        ("http://example.com/a//../.", "http://example.com/a/"),
+        ("http://example.com/a//../..", "http://example.com/"),
+        ("http://example.com/a//../../", "http://example.com/"),
+        ("http://example.com/a//../..//", "http://example.com//"),
+        ("http://example.com/a//..//..", "http://example.com/a/"),
+        ("http://example.com/a//../b", "http://example.com/a/b"),
+        ("http://example.com/a//..//", "http://example.com/a//"),
+        ("http://example.com/a//..//.", "http://example.com/a//"),
+        ("http://example.com/a//..//b", "http://example.com/a//b"),
+        # Normalization with and without an authority component
+        ("memory:/a/b/..", "memory:/a/"),
+        ("memory:/a/b/../..", "memory:/"),
+        ("memory:/a/b/../../..", "memory:/"),
+        ("memory://a/b/..", "memory://a/"),
+        ("memory://a/b/../..", "memory://a/"),
+        ("memory://a/b/../../..", "memory://a/"),
+    ),
+)
+
+
+@pytest.mark.parametrize(*NORMALIZATIONS)
+def test_normalize(unnormalized, normalized):
+    expected = str(UPath(normalized))
+    # Normalise only, do not attempt to follow redirects for http:// paths here
+    result = str(UPath.resolve(UPath(unnormalized)))
+    assert expected == result