Richer AccessorPath API

pathable/accessors.py

@@ -153,7 +153,11 @@ def _read_cached(cls, node_id: int, parts: PDeque[CSK]) -> CSV:
 class LookupAccessor(CachedSubscriptableAccessor[LookupKey, LookupValue]):
 
     def stat(self, parts: Sequence[LookupKey]) -> Union[dict[str, Any], None]:
-        node = self._get_node(self.node, pdeque(parts))
+        try:
+            node = self._get_node(self.node, pdeque(parts))
+        except KeyError:
+            return None
+
         if isinstance(node, Mapping):
             return {
                 'type': 'mapping',
@@ -164,7 +168,15 @@ def stat(self, parts: Sequence[LookupKey]) -> Union[dict[str, Any], None]:
                 'type': 'list',
                 'length': len(node),
             }
-        return None
+        try:
+            length = len(node)
+        except TypeError:
+            length = None
+
+        return {
+            'type': type(node),
+            'length': length,
+        }
 
     def keys(self, parts: Sequence[LookupKey]) -> Sequence[LookupKey]:
         node = self._get_node(self.node, pdeque(parts))

pathable/paths.py

@@ -127,6 +127,17 @@ def __init__(self, accessor: NodeAccessor[N, K, V], *args: Any, separator: Optio
         object.__setattr__(self, 'accessor', accessor)
         super().__init__(*args, separator=separator)
 
+    @classmethod
+    def _from_parts(
+        cls: Type[TAccessorPath],
+        args: Sequence[Any],
+        separator: Optional[str] = None,
+        accessor: Union[NodeAccessor[N, K, V], None] = None,
+    ) -> TAccessorPath:
+        if accessor is None:
+            raise ValueError("accessor must be provided")
+        return cls(accessor, *args, separator=separator)
+
     @classmethod
     def _from_parsed_parts(
         cls: Type[TAccessorPath],
@@ -159,87 +170,68 @@ def _make_child_relpath(
             parts, separator=self.separator, accessor=self.accessor,
         )
 
+    def __rtruediv__(self: TAccessorPath, key: Hashable) -> TAccessorPath:
+        try:
+            return self._from_parts(
+                (key, ) + self.parts,
+                separator=self.separator,
+                accessor=self.accessor,
+            )
+        except TypeError:
+            return NotImplemented
+
+    def __floordiv__(self: TAccessorPath, key: K) -> TAccessorPath:
+        """Return a new existing path with the key appended."""
+        if key not in self:
+            raise KeyError(key)
+        return self / key
+
+    def __rfloordiv__(self: TAccessorPath, key: K) -> TAccessorPath:
+        """Return a new existing path with the key prepended."""
+        new = key / self
+        if not new.exists():
+            raise KeyError(key)
+        return new
+
     def __iter__(self: TAccessorPath) -> Iterator[TAccessorPath]:
+        """Iterate over all child paths."""
         for key in self.accessor.keys(self.parts):
             yield self._make_child_relpath(key)
 
-    def __getitem__(self, key: K) -> Any:
-        if key not in self:
-            raise KeyError(key)
-        path = self / key
+    def __getitem__(self, key: K) -> V:
+        """Access a child path's value."""
+        path = self // key
         return path.read_value()
 
     def __contains__(self, key: K) -> bool:
+        """Check if a key exists in the path."""
         return key in self.accessor.keys(self.parts)
 
     def __len__(self) -> int:
+        """Return the number of child paths."""
         return self.accessor.len(self.parts)
 
+    def exists(self) -> bool:
+        """Check if the path exists."""
+        return self.accessor.stat(self.parts) is not None
+
     def keys(self) -> Sequence[K]:
+        """Return all keys at the current path."""
         return self.accessor.keys(self.parts)
 
-    def getkey(self, key: K, default: Any = None) -> Any:
-        """Return the value for key if key is in the path, else default."""
-        warnings.warn(
-            "'getkey' method is deprecated. Use 'key not in path' and 'path.read_value' instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        try:
-            return self[key]
-        except KeyError:
-            return default
-
-    def iter(self: TAccessorPath) -> Iterator[TAccessorPath]:
-        """Iterate over all child paths."""
-        warnings.warn(
-            "'iter' method is deprecated. Use 'iter(path)' instead.",
-            DeprecationWarning,
-        )
-        return iter(self)
-
-    def iteritems(self: TAccessorPath) -> Iterator[tuple[K, TAccessorPath]]:
-        """Return path's items."""
-        warnings.warn(
-            "'iteritems' method is deprecated. Use 'items' instead.",
-            DeprecationWarning,
-        )
-        return self.items()
-
     def items(self: TAccessorPath) -> Iterator[tuple[K, TAccessorPath]]:
         """Return path's items."""
         for key in self.accessor.keys(self.parts):
             yield key, self._make_child_relpath(key)
 
-    def content(self) -> Any:
-        warnings.warn(
-            "'content' method is deprecated. Use 'read_value' instead.",
-            DeprecationWarning,
-        )
-        return self.read_value()
-
     def get(self, key: K, default: Any = None) -> Any:
-        """Return the child path for key if key is in the path, else default."""
-        warnings.warn(
-            "'get' method is deprecated. Use 'key in path' and 'path / key' instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        if key in self:
-            return self / key
-        return default
-
-    @contextmanager
-    def open(self) -> Any:
-        """Open the path."""
-        warnings.warn(
-            "'open' method is deprecated. Use 'read_value' instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        yield self.read_value()
+        """Return the value for key if key is in the path, else default."""
+        try:
+            return self[key]
+        except KeyError:
+            return default
 
-    def read_value(self) -> Any:
+    def read_value(self) -> V:
         """Return the path's value."""
         return self.accessor.read(self.parts)
 

tests/unit/test_paths.py

@@ -806,6 +806,77 @@ def test_non_existing_key_default_defined(self):
 
         assert result == value
 
+    @pytest.mark.parametrize(
+        "resource,key,expected",
+        (
+            [
+                {"test1": "test2"},
+                "test1",
+                'test2',
+            ],
+            [
+                {"test1": {"test2": "test3"}},
+                "test1",
+                {'test2': 'test3'},
+            ],
+        ),
+    )
+    def test_key_exists(self, resource, key, expected):
+        p = LookupPath._from_lookup(resource)
+
+        result = p.get(key)
+
+        assert result == expected
+
+class TestLookupPathExists:
+    def test_non_existing_key(self):
+        value = "testvalue"
+        resource = {"test1": {"test2": {"test3": value}}}
+        p = LookupPath._from_lookup(resource, "test1/test2/non_existing_key")
+
+        result = p.exists()
+
+        assert result is False
+
+    @pytest.mark.parametrize(
+        "resource,key",
+        (
+            [
+                {"test1": "test2"},
+                "test1",
+            ],
+            [
+                {"test1": 123},
+                "test1",
+            ],
+            [
+                {"test1": True},
+                "test1",
+            ],
+            [
+                {"test1": {"test2": "test3"}},
+                "test1",
+            ],
+        ),
+    )
+    def test_key_exists(self, resource, key):
+        p = LookupPath._from_lookup(resource, key)
+
+        result = p.exists()
+
+        assert result is True
+
+
+class TestLookupPathFloorDiv:
+
+    def test_non_existing_key(self):
+        value = "testvalue"
+        resource = {"test1": {"test2": {"test3": value}}}
+        p = LookupPath._from_lookup(resource, "test1/test2")
+
+        with pytest.raises(KeyError):
+            p // "non_existing_key"
+
     @pytest.mark.parametrize(
         "resource,key,expected",
         (
@@ -826,7 +897,41 @@ def test_non_existing_key_default_defined(self):
     def test_key_exists(self, resource, key, expected):
         p = LookupPath._from_lookup(resource)
 
-        result = p.get(key)
+        result = p // key
+
+        assert result == expected
+
+class TestLookupPathRfloorDiv:
+
+    def test_non_existing_key(self):
+        value = "testvalue"
+        resource = {"test1": {"test2": {"test3": value}}}
+        p = LookupPath._from_lookup(resource, "test1/test2")
+
+        with pytest.raises(KeyError):
+            "non_existing_key" // p
+
+    @pytest.mark.parametrize(
+        "resource,key,expected",
+        (
+            [
+                {"test1": "test2"},
+                "test1",
+                LookupPath._from_lookup({"test1": "test2"}, "test1"),
+            ],
+            [
+                {"test1": {"test2": "test3"}},
+                "test1",
+                LookupPath._from_lookup(
+                    {"test1": {"test2": "test3"}}, "test1"
+                ),
+            ],
+        ),
+    )
+    def test_key_exists(self, resource, key, expected):
+        p = LookupPath._from_lookup(resource)
+
+        result = key // p
 
         assert result == expected