Add local and remote Dandiset directory browsing

dandi/dandiapi.py

@@ -15,7 +15,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable, Iterator, Sequence
 from concurrent.futures import ThreadPoolExecutor
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 from fnmatch import fnmatchcase
@@ -52,7 +52,7 @@
 )
 from .exceptions import HTTP404Error, NotFoundError, SchemaVersionError
 from .keyring_utils import keyring_lookup, keyring_save
-from .misctypes import Digest, RemoteReadableAsset
+from .misctypes import BasePath, Digest, RemoteReadableAsset
 from .utils import (
     USER_AGENT,
     check_dandi_version,
@@ -1426,6 +1426,20 @@ def get_assets(self, order: str | None = None) -> Iterator[RemoteAsset]:
                 f"No such version: {self.version_id!r} of Dandiset {self.identifier}"
             )
 
+    def get_path(self, path: str = "") -> RemoteDandisetPath:
+        """Return a lazy path for browsing this version's asset directories.
+
+        Listing a directory uses the paginated `/assets/paths/` endpoint.
+        Listed children retain their type, recursive count and size, so reading
+        those properties makes no further requests. An arbitrary unlisted path
+        requires a listing of its parent to determine whether it exists.
+
+        Path objects cache listings; call this method again to see later changes.
+
+        .. versionadded:: 0.80.0
+        """
+        return RemoteDandisetPath(parts=(), dandiset=self) / path
+
     def get_asset(self, asset_id: str) -> RemoteAsset:
         """
         Fetch the asset in this version of the Dandiset with the given asset
@@ -2350,3 +2364,120 @@ class ZarrEntryServerData(BaseModel):
     last_modified: datetime = Field(alias="LastModified")
     etag: str = Field(alias="ETag")
     size: int = Field(alias="Size")
+
+
+@dataclass
+class RemoteDandisetPath(BasePath):
+    """A cached view of an asset or virtual directory in a Dandiset version.
+
+    Zarr assets are leaves, just like blob assets. Their internal chunks are not
+    Dandiset children. Retrieving a full asset with `get_asset` makes a separate
+    request because the directory endpoint only supplies its identifier and URL.
+
+    .. versionadded:: 0.80.0
+    """
+
+    #: The Dandiset version containing this path.
+    dandiset: RemoteDandiset
+    _entry: dict[str, Any] | None = field(default=None, repr=False, compare=False)
+    _children: list[RemoteDandisetPath] | None = field(
+        default=None, repr=False, compare=False
+    )
+    _missing: bool = field(default=False, repr=False, compare=False)
+
+    def _get_subpath(self, name: str) -> RemoteDandisetPath:
+        if not name or "/" in name:
+            raise ValueError(f"Invalid path component: {name!r}")
+        if name == ".":
+            return self
+        if name == "..":
+            return self.parent
+        return type(self)(parts=(*self.parts, name), dandiset=self.dandiset)
+
+    @property
+    def parent(self) -> RemoteDandisetPath:
+        return type(self)(parts=self.parts[:-1], dandiset=self.dandiset)
+
+    def _list_entries(self, prefix: str) -> Iterator[dict[str, Any]]:
+        yield from self.dandiset.client.paginate(
+            f"{self.dandiset.version_api_path}assets/paths/",
+            params={"path_prefix": prefix},
+        )
+
+    def _resolve(self) -> dict[str, Any] | None:
+        if self._entry is None and not self._missing:
+            try:
+                if self.is_root():
+                    self._load_children()
+                    self._entry = {
+                        "asset": None,
+                        "aggregate_files": sum(
+                            c.aggregate_files for c in self._children or []
+                        ),
+                        "aggregate_size": sum(c.size for c in self._children or []),
+                    }
+                else:
+                    self._entry = next(
+                        (
+                            e
+                            for e in self._list_entries(str(self.parent))
+                            if e["path"] == str(self)
+                        ),
+                        None,
+                    )
+            except HTTP404Error:
+                self._missing = True
+            if self._entry is None:
+                self._missing = True
+        return self._entry
+
+    def _require_entry(self) -> dict[str, Any]:
+        entry = self._resolve()
+        if entry is None:
+            raise NotFoundError(f"No such Dandiset path: {str(self)!r}")
+        return entry
+
+    def exists(self) -> bool:
+        return self._resolve() is not None
+
+    def is_file(self) -> bool:
+        entry = self._resolve()
+        return entry is not None and entry["asset"] is not None
+
+    def is_dir(self) -> bool:
+        entry = self._resolve()
+        return entry is not None and entry["asset"] is None
+
+    def _load_children(self) -> None:
+        if self._children is None:
+            self._children = [
+                type(self)(
+                    parts=tuple(entry["path"].split("/")),
+                    dandiset=self.dandiset,
+                    _entry=entry,
+                )
+                for entry in self._list_entries(str(self))
+            ]
+
+    def iterdir(self) -> Iterator[RemoteDandisetPath]:
+        if self._require_entry()["asset"] is not None:
+            raise NotADirectoryError(str(self))
+        self._load_children()
+        yield from self._children or []
+
+    @property
+    def aggregate_files(self) -> int:
+        """The recursive number of assets under this path (one for an asset)."""
+        return int(self._require_entry()["aggregate_files"])
+
+    @property
+    def size(self) -> int:
+        """The recursive size in bytes, as reported by the Archive."""
+        return int(self._require_entry()["aggregate_size"])
+
+    def get_asset(self) -> RemoteAsset:
+        """Fetch the full asset record; directories raise IsADirectoryError."""
+        asset = self._require_entry()["asset"]
+        if asset is None:
+            raise IsADirectoryError(str(self))
+        return self.dandiset.get_asset(asset["asset_id"])

dandi/dandiset.py

@@ -1,4 +1,5 @@
 """Classes/utilities for support of a dandiset"""
+
 from __future__ import annotations
 
 from collections.abc import Iterable, Iterator
@@ -11,7 +12,9 @@
 
 from . import get_logger
 from .consts import dandiset_metadata_file
+from .exceptions import NotFoundError
 from .files import DandisetMetadataFile, LocalAsset, dandi_file, find_dandi_files
+from .misctypes import BasePath
 from .utils import find_parent_directory_containing, under_paths, yaml_dump, yaml_load
 
 if TYPE_CHECKING:
@@ -168,6 +171,18 @@ def assets(self, allow_all: bool = False) -> AssetView:
             data[PurePosixPath(df.path)] = df
         return AssetView(data)
 
+    def get_path(self, path: str = "") -> LocalDandisetPath:
+        """Browse a snapshot of the Dandiset's discoverable assets.
+
+        Discovery runs once for the whole Dandiset and includes generic assets.
+        Empty and hidden directories are excluded by normal discovery rules;
+        Zarr directories are represented as single assets. Descendants share
+        the snapshot. Call this method again to refresh it.
+
+        .. versionadded:: 0.80.0
+        """
+        return LocalDandisetPath(parts=(), assets=self.assets(allow_all=True)) / path
+
     def metadata_file(self) -> DandisetMetadataFile:
         df = dandi_file(self._metadata_file_obj, dandiset_path=self.path)
         assert isinstance(df, DandisetMetadataFile)
@@ -192,3 +207,70 @@ def under_paths(self, paths: Iterable[str | PurePath]) -> Iterator[LocalAsset]:
         # contain '.' or '..'
         for p in under_paths(self.data.keys(), paths):
             yield self.data[p]
+
+
+@dataclass
+class LocalDandisetPath(BasePath):
+    """An asset or directory in a local Dandiset discovery snapshot.
+
+    .. versionadded:: 0.80.0
+    """
+
+    #: Shared discovery results, including generic assets.
+    assets: AssetView
+
+    def _get_subpath(self, name: str) -> LocalDandisetPath:
+        if not name or "/" in name:
+            raise ValueError(f"Invalid path component: {name!r}")
+        if name == ".":
+            return self
+        if name == "..":
+            return self.parent
+        return type(self)(parts=(*self.parts, name), assets=self.assets)
+
+    @property
+    def parent(self) -> LocalDandisetPath:
+        return type(self)(parts=self.parts[:-1], assets=self.assets)
+
+    def _descendants(self) -> Iterator[LocalAsset]:
+        yield from self.assets.under_paths([PurePosixPath(str(self))])
+
+    def exists(self) -> bool:
+        return self.is_root() or next(self._descendants(), None) is not None
+
+    def is_file(self) -> bool:
+        return PurePosixPath(str(self)) in self.assets.data
+
+    def is_dir(self) -> bool:
+        return self.exists() and not self.is_file()
+
+    def iterdir(self) -> Iterator[LocalDandisetPath]:
+        if not self.exists():
+            raise NotFoundError(f"No such Dandiset path: {str(self)!r}")
+        if self.is_file():
+            raise NotADirectoryError(str(self))
+        names = {a.path.split("/")[len(self.parts)] for a in self._descendants()}
+        for name in sorted(names):
+            yield self / name
+
+    @property
+    def aggregate_files(self) -> int:
+        """The recursive number of discoverable assets."""
+        if not self.exists():
+            raise NotFoundError(f"No such Dandiset path: {str(self)!r}")
+        return sum(1 for _ in self._descendants())
+
+    @property
+    def size(self) -> int:
+        """The total size in bytes of the assets below this path."""
+        if not self.exists():
+            raise NotFoundError(f"No such Dandiset path: {str(self)!r}")
+        return sum(a.size for a in self._descendants())
+
+    def get_asset(self) -> LocalAsset:
+        """Return the discovered asset; directories raise IsADirectoryError."""
+        if not self.exists():
+            raise NotFoundError(f"No such Dandiset path: {str(self)!r}")
+        if not self.is_file():
+            raise IsADirectoryError(str(self))
+        return self.assets.data[PurePosixPath(str(self))]