feat(client)!: read releases through the coordinator

packages/timenet-connectors/src/timenet_connectors/api.py

@@ -111,9 +111,9 @@ def load(
 
     Args:
         dataset_id: The dataset ID (``org/name``).
-        version: The version declared by the connector, or ``None`` to load the latest version.
-        fetch_from: The registry, or its URI, to copy missing parents and payload sources from.
-        build_parents: Build missing parent releases from their installed connectors.
+        version: The version to load, which the card must declare, or ``None`` for the latest.
+        fetch_from: The registry or URI that supplies missing parents and payload sources.
+        build_parents: Build missing parent releases with installed connectors.
 
     Returns:
         The loaded dataset.

packages/timenet-connectors/tests/test_build_plan.py

@@ -567,22 +567,23 @@ def test_load_obtains_a_missing_parent_before_it_builds_and_reads(tmp_path, monk
     assert calls == (builds if obtain == "fetch_from" else [(PARENT.dataset_id, False), *builds])
 
 
-def test_load_copies_a_missing_payload_source_through_fetch_from(tmp_path, monkeypatch):
+def test_load_reads_a_missing_payload_source_through_fetch_from(tmp_path, monkeypatch):
     out = LocalRegistry(tmp_path / "out")
     _parent_without_its_source(out)
     origin = LocalRegistry(tmp_path / "origin")
     origin.put_dataset(make_dataset())
     monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
     monkeypatch.setenv("TIMENET_REGISTRY", str(out.root))
     # The build has nothing to do, and no builder backend can build the payload source, so only the
-    # read can copy it.
+    # read can get it.
     monkeypatch.setattr(api_module, "build", lambda dataset_id, **options: None)
     monkeypatch.setattr(builders, "entry_points", lambda group: [])
 
     dataset = timenet_connectors.load(PARENT.dataset_id, fetch_from=origin)
 
-    assert out.exists(GRAND.dataset_id, str(GRAND.version))
     assert len(dataset.records) == 2
+    # The read opens the payload source in origin, where it is, and copies nothing.
+    assert not out.exists(GRAND.dataset_id, str(GRAND.version))
 
 
 def test_a_pinned_parent_build_hands_its_pins_to_the_isolated_child(tmp_path, monkeypatch):

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

@@ -1,26 +1,24 @@
 """Plan and run release builds with dependencies first."""
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from typing import assert_never
 
 from timenet.builders import BuildOptions, declared
 from timenet.coordinator.plan import BuildStep, CopyStep, Plan, ReadStep, plan_steps
-from timenet.manifest import PayloadSource
 from timenet.progress import StepFinished, StepStarted
 from timenet.registry import copy_release
 from timenet.registry.dependencies import prepare_local_copy
 from timenet.registry.local import LocalRegistry
 from timenet.sources import ReleaseSource
-from timenet.types.references import DatasetRef, is_latest, split_ref
+from timenet.types.references import is_latest, split_ref
 
 
-def plan_release_build(  # ruff: ignore[too-many-arguments]
+def plan_release_build(
     dataset_id: str,
     *,
     store: LocalRegistry,
     parents_from: Sequence[ReleaseSource] = (),
     force: bool = False,
-    pinned_by: Mapping[DatasetRef, PayloadSource] | None = None,
     open_access_only: bool = False,
 ) -> Plan:
     """Plan a release build in a local registry.
@@ -33,7 +31,6 @@ def plan_release_build(  # ruff: ignore[too-many-arguments]
         store: The local registry to write to.
         parents_from: Sources to copy or build missing dependencies from, in preference order.
         force: Plan a forced rebuild and its parents even if the requested release already exists.
-        pinned_by: File pins grouped by dependent release that the build must preserve.
         open_access_only: Require ``access: open`` for every needed release.
 
     Returns:
@@ -52,7 +49,6 @@ def plan_release_build(  # ruff: ignore[too-many-arguments]
         parents_from=parents_from,
         force_build=force,
         open_access_only=open_access_only,
-        pinned_by=pinned_by,
     )
     return Plan(tuple(steps))
 

packages/timenet/src/timenet/coordinator/plan.py

@@ -566,14 +566,13 @@ def _check_agreement(release: DatasetRef, readers: Mapping[DatasetRef, PayloadSo
                 )
 
 
-def plan_steps(  # ruff: ignore[too-many-arguments]
+def plan_steps(
     builder: InstalledBuilder,
     output: LocalRegistry,
     *,
     parents_from: Sequence[ReleaseSource],
     force_build: bool,
     open_access_only: bool,
-    pinned_by: Mapping[DatasetRef, PayloadSource] | None = None,
 ) -> tuple[DatasetRef, list[Step]]:
     """Plan a dataset build and all required releases.
 
@@ -585,7 +584,6 @@ def plan_steps(  # ruff: ignore[too-many-arguments]
         parents_from: The sources to search for missing releases, in order.
         force_build: Rebuild the requested release even if the output holds it.
         open_access_only: Require open access for every release.
-        pinned_by: The releases that pin files of the requested release, with their file requirements.
 
     Returns:
         The requested release and ordered steps, ending with its build if needed.
@@ -596,8 +594,6 @@ def plan_steps(  # ruff: ignore[too-many-arguments]
     root = builder.release
     # The graph leaves the output out of the sources of its copies and builds.
     graph = _Graph(output, parents_from, open_only=open_access_only)
-    if pinned_by:
-        graph.pins[root] = dict(pinned_by)
     # If the output already holds the requested release, a reader needs only its payload sources to
     # open it. Its declared parents matter only to a rebuild.
     held, damage = (None, None) if force_build else graph.in_output(root)

packages/timenet/src/timenet/registry/dependencies.py

@@ -272,7 +272,6 @@ def prepare_local_copy(  # ruff: ignore[too-many-arguments]
     manifest: Manifest | None = None,
     force: bool = False,
     progress_cb: Callable[[int], None] | None = None,
-    before_fetch: Callable[[Manifest], object] | None = None,
 ) -> DatasetVersion:
     """Ensure complete local copies of a release and of each of its payload sources.
 
@@ -294,9 +293,6 @@ def prepare_local_copy(  # ruff: ignore[too-many-arguments]
         manifest: An already-loaded manifest for the requested release.
         force: Fetch the requested release again, even if it is cached.
         progress_cb: Report downloaded bytes.
-        before_fetch: A callback that gets the manifest of each release before this function
-            downloads the files of the release. For example, the callback can check the access
-            requirement of the release.
 
     Returns:
         The requested release with its payload sources attached and checked against its pins.
@@ -319,7 +315,6 @@ def open_complete(ref: DatasetRef, *, requested: bool) -> DatasetVersion:
             manifest=manifest if requested else None,
             force=force and requested,
             progress_cb=progress_cb,
-            before_fetch=before_fetch,
         )
         return DatasetVersion(release, pafs.LocalFileSystem(), str(directory))
 
@@ -362,16 +357,17 @@ def _ensure_dataset_version_available_locally(  # ruff: ignore[too-many-argument
     manifest: Manifest | None,
     force: bool,
     progress_cb: Callable[[int], None] | None,
-    before_fetch: Callable[[Manifest], object] | None = None,
 ) -> Manifest:
-    """Reuse or fetch one complete release, and repair a damaged copy only by fetching it.
+    """Reuse a complete local release or fetch it from a registry.
+
+    Local checks compare file sizes. Downloads also check file hashes.
 
     Returns:
-        The release manifest. File hashes are checked during transfer, not on every read.
+        The complete release's manifest.
 
     Raises:
-        TimeNetDatasetNotFoundError: If the release is absent and this function cannot fetch it.
-        TimeNetFileError: If a local file is damaged and ``fetch_from`` is not given, or a fetched file does not match.
+        TimeNetDatasetNotFoundError: If the release is missing and ``fetch_from`` is unset or lacks it.
+        TimeNetFileError: If a local file is damaged and ``fetch_from`` is unset, or a downloaded file is damaged.
         TimeFFormatError: If a manifest identifies a different release.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the helpers
     if manifest is not None:
@@ -391,9 +387,7 @@ def _ensure_dataset_version_available_locally(  # ruff: ignore[too-many-argument
         else:
             return local
     if fetch_from is not None:
-        return _download_dataset_version(
-            directory, reference, fetch_from, manifest=manifest, progress_cb=progress_cb, before_fetch=before_fetch
-        )
+        return _download_dataset_version(directory, reference, fetch_from, manifest=manifest, progress_cb=progress_cb)
     raise TimeNetDatasetNotFoundError(f"complete release {reference} is missing at {directory}")
 
 
@@ -417,27 +411,22 @@ def _load_complete_local_manifest(
     return manifest if complete_copy(directory, manifest) else None
 
 
-def _download_dataset_version(  # ruff: ignore[too-many-arguments]
+def _download_dataset_version(
     directory: Path,
     reference: DatasetRef,
     registry: "BaseRegistry",
     *,
     manifest: Manifest | None,
     progress_cb: Callable[[int], None] | None,
-    before_fetch: Callable[[Manifest], object] | None = None,
 ) -> Manifest:
-    """Check a release's identity and download its files.
-
-    Run ``before_fetch`` after checking the manifest and before downloading files.
+    """Download a complete release after checking its dataset ID and version.
 
     Returns:
-        The manifest of the downloaded and verified release.
+        The manifest for the downloaded and verified release.
     """
     if manifest is None:
         manifest = registry.get_manifest(reference.dataset_id, str(reference.version))
     validate_manifest_reference(manifest, reference)
-    if before_fetch is not None:
-        before_fetch(manifest)
     registry.download_release(reference, directory, force=True, manifest=manifest, progress_cb=progress_cb)
     return manifest
 

packages/timenet/tests/coordinator/test_build.py

@@ -119,20 +119,6 @@ def _store_child_without_parent(registry):
     return registry.get_manifest(CHILD_REF.dataset_id).payload_sources[SourceKey(str(PARENT_REF))]
 
 
-@pytest.fixture
-def pins(store):
-    """Store the child without the parent, and return the pins of the child on the files of the parent."""
-    return _store_child_without_parent(store)
-
-
-def test_a_build_passes_the_pins_of_the_caller_to_the_backend(store, fake_parent, pins):
-    plan = plan_release_build(PARENT_REF.dataset_id, store=store, pinned_by={CHILD_REF: pins})
-    run_plan(plan, store)
-    assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})]
-    assert plan.steps == (_build(PARENT_REF, pinned_by={CHILD_REF: pins}),)
-    assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version))
-
-
 def test_plan_release_build_rejects_a_version_the_connector_does_not_declare(store, fake):
     with pytest.raises(TimeNetVersionUnavailableError) as caught:
         plan_release_build("a/b@2.0.0", store=store)

packages/timenet/tests/registry/test_remote.py

@@ -578,7 +578,7 @@ def _gated_release(origin: Path, page: str) -> tuple[str, DatasetRef]:
 
 
 @pytest.mark.parametrize("page", ["own", "ancestor"])
-def test_a_local_client_checks_access_before_it_copies_from_a_remote_registry(tmp_path, page):
+def test_a_local_client_checks_access_before_it_reads_from_a_remote_registry(tmp_path, page):
     dataset_id, release = _gated_release(tmp_path / "origin", page)
     transport, requests = build_fake(next((tmp_path / "origin").rglob("manifest.json")).parent)
     remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache")
@@ -587,6 +587,7 @@ def test_a_local_client_checks_access_before_it_copies_from_a_remote_registry(tm
         TimeNet(tmp_path / "local").open(dataset_id, auto_build=False, fetch_from=remote)
 
     assert caught.value.requirements == (AccessRequirement(release, Access.CREDENTIALED, "https://example.org/dua"),)
+    assert caught.value.hint == "Build it from its connector with auto_build=True."
     assert not any("/download/" in request.url.path or request.url.host == "blob.local" for request in requests)
     assert not (tmp_path / "local").exists()