feat(sources)!: resolve a release through an ordered list of sources

packages/timenet-connectors/src/timenet_connectors/builder/cli.py

@@ -33,7 +33,7 @@
     open_writable_registry,
 )
 from timenet.registry._paths import release_dir  # ruff: ignore[import-private-name]
-from timenet.types.references import DatasetRef, split_ref
+from timenet.types.references import DatasetRef, is_latest, split_ref
 from timenet_connectors.builder.backend import ConnectorBuilder
 from timenet_connectors.discovery import known_ids
 
@@ -141,7 +141,7 @@ def build(
     known = known_ids()
     if dataset_id not in known:
         raise typer.BadParameter(f"no connector for {dataset_id!r}; known: {', '.join(known) or '(none)'}")
-    if version is not None:
+    if not is_latest(version):
         declared(dataset_id, version)
     try:
         # The parent shows each copy and build step. An isolated child shows its own downloads and shards.

packages/timenet-connectors/tests/test_build_backend.py

@@ -16,6 +16,8 @@
 from timenet.progress import BytesMoved, ShardWritten
 from timenet.reader import TimeFReader
 from timenet.registry import LocalRegistry, open_local_version
+from timenet.sources import FoundInConnector, InstalledConnectors
+from timenet.types import Version
 from timenet.types.references import DatasetRef
 from timenet.values_backends import ValuesBackend
 from timenet.writer.progress import ProgressStage, WriteProgressEvent
@@ -99,6 +101,11 @@ def test_a_card_that_cannot_be_read_is_skipped_with_a_warning(tmp_path, monkeypa
         found = installed_builder("timenet/hello-world")
     assert found is not None
     assert found.card.dataset_id == "timenet/hello-world"
+    # The installed connectors still answer for the other datasets.
+    with pytest.warns(ConnectorCardWarning, match=expected):
+        answer = InstalledConnectors().resolve("timenet/hello-world", None)
+    assert isinstance(answer, FoundInConnector)
+    assert answer.version == Version(1, 0, 0)
 
 
 def test_build_in_process_round_trips_the_connector_default_zarr_backend(tmp_path, monkeypatch):

packages/timenet/src/timenet/coordinator/build.py

@@ -13,7 +13,7 @@
 from timenet.registry.dependencies import prepare_local_copy
 from timenet.registry.local import LocalRegistry
 from timenet.registry.shared import complete_copy
-from timenet.types.references import DatasetRef, split_ref
+from timenet.types.references import DatasetRef, is_latest, split_ref
 
 
 def build_release(  # ruff: ignore[too-many-arguments]
@@ -59,7 +59,8 @@ def build_release(  # ruff: ignore[too-many-arguments]
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the plan, the copy and the backends
     options = options or BuildOptions()
     dataset_id, version = split_ref(dataset)
-    found = declared(dataset_id, version)
+    # A connector builds only the version of its card, so "latest" asks for that version.
+    found = declared(dataset_id, None if is_latest(version) else version)
     card, requested = found.card, found.release
     steps = plan_parents(
         dataset_id,

packages/timenet/src/timenet/registry/base.py

@@ -21,7 +21,9 @@
 from timenet.registry.dependencies import prepare_local_copy, validate_manifest_reference
 from timenet.registry.shared import complete_copy
 from timenet.registry.version import DatasetVersion
-from timenet.types import DatasetMetadata, Domain, License, Task
+from timenet.sources import FoundInRegistry, NotFound, NotServed, ReleaseLookup
+from timenet.types import Access, DatasetMetadata, Domain, License, Task, Version
+from timenet.types.access import access_requirements, effective_access
 from timenet.types.references import DatasetRef
 
 
@@ -33,7 +35,7 @@
 
 
 class BaseRegistry(ABC):
-    """A registry of dataset manifests, files, and searchable metadata."""
+    """A registry that provides TimeF dataset metadata, manifests, and files."""
 
     def __str__(self) -> str:
         return type(self).__name__
@@ -127,6 +129,29 @@ def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVe
             TimeNetDatasetNotFoundError: If the dataset id or version is unknown.
         """
 
+    def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup:
+        """Check whether the registry serves a dataset release.
+
+        If :attr:`requires_open_access` is true, the registry serves a release only if its card and each card in its
+        lineage declare open access.
+
+        Args:
+            dataset_id: The dataset ID.
+            version: An exact version, or ``None`` for the latest version.
+
+        Returns:
+            ``FoundInRegistry`` if served, ``NotServed`` if access is restricted, or ``NotFound`` if absent.
+        """
+        try:
+            manifest = self.get_manifest(dataset_id, None if version is None else str(version))
+        except TimeNetDatasetNotFoundError:
+            return NotFound()
+        held = manifest.metadata.dataset_version
+        access = effective_access(manifest.metadata, manifest.lineage)
+        if access is Access.OPEN or not self.requires_open_access:
+            return FoundInRegistry(held, self, manifest)
+        return NotServed(held, access, access_requirements(manifest.metadata, manifest.lineage), manifest)
+
     def download_version(  # ruff: ignore[too-many-arguments]
         self,
         dataset_id: str,

packages/timenet/src/timenet/registry/local.py

@@ -9,21 +9,22 @@
 
 from timenet._fs import staged_directory
 from timenet.dataset import TimeFDataset
-from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetFileError
+from timenet.errors import TimeFFormatError, TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError
 from timenet.format.constants import MANIFEST_FILE
 from timenet.manifest import Manifest, PayloadSource, read_manifest
 from timenet.registry._paths import release_dir
 from timenet.registry.dependencies import check_file_sizes, open_local_release
 from timenet.registry.version import DatasetVersion
 from timenet.registry.writable import WritableRegistry
+from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup
 from timenet.types import DatasetMetadata, Version
 from timenet.types.metadata_fields import validate_dataset_id
 from timenet.types.references import DatasetRef, is_latest
 from timenet.writer import TimeFWriter, WriteProgressEvent
 
 
 class LocalRegistry(WritableRegistry):
-    """Read and publish datasets in a local directory."""
+    """A registry that reads and writes datasets in a local directory."""
 
     def __init__(self, root: Path) -> None:
         """Open a local registry rooted at a directory.
@@ -123,6 +124,33 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest:
         self._manifest_cache[dataset_id, resolved] = (mtime, manifest)
         return manifest
 
+    def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup:
+        """Check whether the local registry holds a dataset release.
+
+        Local releases are available at any access level.
+
+        Args:
+            dataset_id: The dataset ID.
+            version: An exact version, or ``None`` for the latest version.
+
+        Returns:
+            ``FoundInRegistry`` with the committed version, or ``NotFound`` if the release is absent.
+
+        Raises:
+            TimeFValidationError: If ``dataset_id`` is invalid.
+        """
+        try:
+            validate_dataset_id(dataset_id)
+        except ValueError as error:
+            raise TimeFValidationError(f"invalid dataset ID {dataset_id!r}: {error}") from error
+        resolved = self._latest_version(dataset_id) if version is None else str(version)
+        if resolved is None:
+            return NotFound()
+        reference = DatasetRef.parse(f"{dataset_id}@{resolved}")
+        if not (release_dir(self._root, reference) / MANIFEST_FILE).exists():
+            return NotFound()
+        return FoundInRegistry(reference.version, self)
+
     def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO:
         """Open one file of a dataset version for binary reading.
 

packages/timenet/src/timenet/registry/remote/_http.py

@@ -2,13 +2,14 @@
 
 Owns how the SDK talks to the hosted registry. It prefixes ``/api/v1`` and attaches the bearer token
 and a ``timenet/<version>`` User-Agent to API calls. It resolves presigned download URLs without
-leaking the token to object storage. It maps HTTP status codes onto TimeNet errors. A sync
-:class:`httpx.Client` serves the control plane. :meth:`new_async_client` hands out an
+leaking the token to object storage. It maps HTTP status codes and network failures onto TimeNet
+errors. A sync :class:`httpx.Client` serves the control plane. :meth:`new_async_client` hands out an
 :class:`httpx.AsyncClient` sharing the same configuration for the parallel downloader.
 """
 
 import asyncio
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Generator, Mapping
+from contextlib import contextmanager
 import importlib.metadata
 import logging
 import math
@@ -124,6 +125,22 @@ async def _send_with_retry_async(send: Callable[[], Awaitable[httpx.Response]])
     return await send()
 
 
+@contextmanager
+def _reaching(where: str) -> Generator[None]:
+    """Name ``where`` in the error of a request that fails on the network.
+
+    Args:
+        where: The registry or storage location to name in the error.
+
+    Raises:
+        TimeNetRegistryError: If a request fails to reach ``where``.
+    """
+    try:
+        yield
+    except httpx.TransportError as error:
+        raise TimeNetRegistryError(f"cannot reach {where}: {error}") from error
+
+
 def timenet_user_agent() -> str:
     """Return the ``timenet/<version>`` User-Agent used for backend tracing.
 
@@ -192,7 +209,7 @@ def _redirect_target(response: httpx.Response, path: str) -> str:
 
 
 class RegistryHttpClient:
-    """A thin httpx wrapper for the registry API and its presigned object URLs."""
+    """An HTTP client for the registry API and signed file URLs."""
 
     @property
     def token(self) -> str | None:
@@ -253,15 +270,16 @@ def api_headers(self) -> dict[str, str]:
         return headers
 
     def get_text(self, path: str) -> str:
-        """GET an API path and return the raw text body (used for ``manifest.json``).
+        """Fetch the text body from a registry API path.
 
         Args:
             path: A path under ``/api/v1``.
 
         Returns:
             The response text.
         """
-        response = _send_with_retry(lambda: self._client.get(f"{API_PREFIX}{path}", headers=self.api_headers()))
+        with _reaching(self._base_url):
+            response = _send_with_retry(lambda: self._client.get(f"{API_PREFIX}{path}", headers=self.api_headers()))
         _raise_for_status(response)
         return response.text
 
@@ -270,20 +288,21 @@ def post(self, path: str, *, content: bytes | None = None, json: Any = None) ->
 
         Args:
             path: A path under ``/api/v1``.
-            content: Raw request body bytes (e.g. the manifest), mutually exclusive with ``json``.
+            content: The request bytes, mutually exclusive with ``json``.
             json: A JSON-serializable body.
 
         Returns:
             The raw response (already status-checked).
         """
-        response = _send_with_retry(
-            lambda: self._client.post(f"{API_PREFIX}{path}", content=content, json=json, headers=self.api_headers())
-        )
+        with _reaching(self._base_url):
+            response = _send_with_retry(
+                lambda: self._client.post(f"{API_PREFIX}{path}", content=content, json=json, headers=self.api_headers())
+            )
         _raise_for_status(response)
         return response
 
     def resolve_presigned(self, dataset_id: str, version: str, relpath: str) -> str:
-        """Resolve a file's presigned URL via the download redirect, without following it.
+        """Get a signed download URL for a dataset file.
 
         Args:
             dataset_id: The ``org/name`` id.
@@ -297,7 +316,10 @@ def resolve_presigned(self, dataset_id: str, version: str, relpath: str) -> str:
             TimeNetRegistryError: If the download endpoint does not redirect.
         """  # ruff: ignore[docstring-extraneous-exception] - raised by _redirect_target
         path = _download_path(dataset_id, version, relpath)
-        response = _send_with_retry(lambda: self._client.get(path, headers=self.api_headers(), follow_redirects=False))
+        with _reaching(self._base_url):
+            response = _send_with_retry(
+                lambda: self._client.get(path, headers=self.api_headers(), follow_redirects=False)
+            )
         return _redirect_target(response, path)
 
     def stream_to(self, url: str, sink: BinaryIO) -> None:
@@ -307,7 +329,7 @@ def stream_to(self, url: str, sink: BinaryIO) -> None:
             url: The presigned URL.
             sink: A writable binary file object.
         """
-        with self._client.stream("GET", url) as response:
+        with _reaching(f"the storage of {self._base_url}"), self._client.stream("GET", url) as response:
             if response.is_error:
                 response.read()  # a streaming response has no body until read; _raise_for_status needs it
                 _raise_for_status(response)
@@ -322,7 +344,8 @@ def put(self, url: str, *, content: bytes, headers: Mapping[str, str]) -> None:
             content: The request body.
             headers: Headers the presign requires (e.g. the checksum binding).
         """
-        response = self._client.put(url, content=content, headers=dict(headers))
+        with _reaching(f"the storage of {self._base_url}"):
+            response = self._client.put(url, content=content, headers=dict(headers))
         _raise_for_status(response)
 
     def new_async_client(self) -> httpx.AsyncClient:

packages/timenet/src/timenet/registry/remote/registry.py

@@ -18,7 +18,7 @@
 from timenet.manifest import Manifest, parse_manifest
 from timenet.registry.lazy_read import LazyReadRegistry
 from timenet.registry.remote._download import ProgressCallback, download_version_files
-from timenet.registry.remote._http import RegistryHttpClient
+from timenet.registry.remote._http import RegistryHttpClient, _reaching
 from timenet.registry.remote.filesystem import RegistryFileSystem
 from timenet.registry.remote.models import (
     DatasetDetailModel,
@@ -151,9 +151,10 @@ def download_release(
         dataset_id, version = reference.dataset_id, str(reference.version)
         if manifest is None:
             manifest = self.get_manifest(dataset_id, version)
-        download_version_files(
-            self._http, manifest, dataset_id, version, Path(directory), force=force, progress_cb=progress_cb
-        )
+        with _reaching(self._base_url):
+            download_version_files(
+                self._http, manifest, dataset_id, version, Path(directory), force=force, progress_cb=progress_cb
+            )
 
     def put_release(self, from_directory: Path) -> DatasetRef:
         """Publish a release directory to the remote registry.