From 2b7f0c64083b11a15e10f8c4c72b3ffc7ea16965 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 12:35:03 +0100 Subject: [PATCH 1/6] feat(coordinator)!: plan parents through an ordered list of sources A build gets each parent that the store lacks from one ordered list of sources: registries to copy from, and InstalledConnectors() to build with. Before this change, the planner took one registry (fetch_from) and one flag (build_parents). The planner now takes parents_from. For each release that the store lacks or holds damaged, resolve_release() picks the first source that provides its exact version. A registry gives a CopyStep from that registry, and InstalledConnectors() gives a BuildStep with the InstalledBuilder of the lookup. Step is the union of the two types. A registry that lists a release but does not serve its files does not count, so a later source can provide the release. A release needs exactly the releases that its manifest lists as payload sources, never the payload sources of a payload source. The copy of a parent brings each of its payload sources that the registry serves. The planner finds them before any step runs, with find_payload_sources(), the walk and access check that copy_release() uses too. A payload source that both the registry and the store lack gets its own step, from its own source, and the copy waits for it. So a copy that cannot be completed fails before any download. A copy of a payload source alone takes only its own files (with_payload_sources=False). After the walk, the planner checks the pins of each pinned release: a copied release against its registry, a held release against the store, and for a release to build, that its readers pin each file alike. So these errors stop the plan before any step runs. In one plan, the planner asks each source at most once for each release, and it keeps each answer. It takes the manifests from the answers, which a registry that is not local fills. It reads a manifest itself only from the store and from a local registry. So the hosted registry gets one request for each manifest. A BuildStep holds the InstalledBuilder that the lookup read, and run_plan() builds with it. So a plan and its run read the card of each connector once. plan_steps() takes the builder of the requested release. The planner now makes the build step of the requested release too. A step waits for the nearest steps that it depends on, also through a release without a step. For example, if the store holds a parent but not its payload source, the build waits for the copy of that payload source. The store is never a source for itself. The planner drops it from the list, also as another LocalRegistry over the same directory, so it never copies the store onto itself. A damaged copy of a parent stops the plan if no other registry provides it. A damaged copy of the requested release gets a forced build, so the plan never ends with "Nothing to do" for it. build_release() and plan_parents() go away. plan_release_build() plans a build, and run_plan() runs the plan with one arm for each step type. plan_release_build() checks the connector and the version, and then it plans. Each BuildStep has a force flag. plan_release_build(force=True) sets it on the build of the requested release, and run_plan() passes it to the backend. run_plan() copies only through copy_release(). The CLI, timenet_connectors.build() and the client keep their keywords for now. Each maps them onto parents_from: the registry of --fetch-from first, then InstalledConnectors() if --build-parents is set. The client always adds InstalledConnectors(). The hint of a missing parent names only the options that the build did not use. If a payload source is missing, copy_release() now names both registries. BREAKING CHANGE: A parent that no installed connector builds at its exact version raises TimeNetMissingParentError, not TimeFValidationError. A parent that a registry lists but does not serve now builds when the build may build parents: with --build-parents, build_parents=True, or the auto_build of the client. Part of TNET-157. --- .../src/timenet_connectors/api.py | 41 +- .../src/timenet_connectors/builder/cli.py | 37 +- .../tests/test_build_plan.py | 374 +++++++---- packages/timenet/src/timenet/client.py | 33 +- .../timenet/src/timenet/coordinator/build.py | 166 ++--- .../timenet/src/timenet/coordinator/plan.py | 595 +++++++++++------- .../timenet/src/timenet/registry/_copy.py | 8 +- packages/timenet/src/timenet/sources.py | 39 +- .../timenet/tests/coordinator/test_build.py | 193 +++--- packages/timenet/tests/registry/test_copy.py | 10 +- packages/timenet/tests/test_client.py | 21 +- 11 files changed, 891 insertions(+), 626 deletions(-) diff --git a/packages/timenet-connectors/src/timenet_connectors/api.py b/packages/timenet-connectors/src/timenet_connectors/api.py index bb2458733..346ce2c43 100644 --- a/packages/timenet-connectors/src/timenet_connectors/api.py +++ b/packages/timenet-connectors/src/timenet_connectors/api.py @@ -24,7 +24,7 @@ def build( # ruff: ignore[too-many-arguments] fetch_from: "str | BaseRegistry | None" = None, build_parents: bool = False, ) -> Path: - # ruff: ignore[docstring-extraneous-exception] (raised by build_release, not directly here) + # ruff: ignore[docstring-extraneous-exception] (raised by the plan and its run, not directly here) """Build a dataset release in a local registry. The build reuses an existing release unless you set ``force``. The connector runs in an environment @@ -48,7 +48,7 @@ def build( # ruff: ignore[too-many-arguments] TimeNetMissingParentError: If a required parent release is still missing. """ from timenet.builders import BuildOptions, declared # ruff: ignore[import-outside-top-level] - from timenet.coordinator.build import build_release # ruff: ignore[import-outside-top-level] + from timenet.coordinator.build import plan_release_build, run_plan # ruff: ignore[import-outside-top-level] from timenet.errors import TimeNetMissingParentError # ruff: ignore[import-outside-top-level] from timenet.registry import ( # ruff: ignore[import-outside-top-level] LocalRegistry, @@ -56,24 +56,49 @@ def build( # ruff: ignore[too-many-arguments] open_registry, ) from timenet.registry._paths import release_dir # ruff: ignore[import-outside-top-level, import-private-name] + from timenet.sources import InstalledConnectors # ruff: ignore[import-outside-top-level] release = declared(dataset_id).release root = Path(out).expanduser() if out is not None else default_registry_path() + source = open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from + store = LocalRegistry(root) try: - build_release( + plan = plan_release_build( f"{dataset_id}@{version}" if version else dataset_id, - store=LocalRegistry(root), - fetch_from=open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from, - build_parents=build_parents, + store=store, + # Maps fetch_from and build_parents onto a list of sources for the coordinator. + parents_from=[ + *([source] if source is not None else []), + *([InstalledConnectors()] if build_parents else []), + ], force=force, - options=BuildOptions(keep_cache=keep_cache), ) + run_plan(plan, store, options=BuildOptions(keep_cache=keep_cache)) except TimeNetMissingParentError as error: - error.hint = "Copy it with fetch_from= or build it with build_parents=True." + error.hint = _missing_hint(fetch_from=fetch_from, build_parents=build_parents) raise return release_dir(root, release) +def _missing_hint(*, fetch_from: "str | BaseRegistry | None", build_parents: bool) -> str | None: + """Name the keywords of :func:`build` that can get a missing release and that the call did not use. + + Args: + fetch_from: The value of ``fetch_from``, or ``None``. + build_parents: The value of ``build_parents``. + + Returns: + The hint as a full sentence, or ``None`` if the call used both keywords. + """ + if fetch_from is None and not build_parents: + return "Copy it with fetch_from= or build it with build_parents=True." + if fetch_from is None: + return "Copy it with fetch_from=." + if not build_parents: + return "Build it with build_parents=True." + return None + + def load( dataset_id: str, version: str | None = None, diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index 7b43b2d7f..fd8fd0bb5 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -15,7 +15,7 @@ from timenet.cli.runner import run_cli from timenet.cli.ui import console, progress_renderer from timenet.config import settings -from timenet.coordinator.build import build_release +from timenet.coordinator.build import plan_release_build, run_plan from timenet.errors import ( TimeNetAccessError, TimeNetFileError, @@ -33,6 +33,7 @@ open_writable_registry, ) from timenet.registry._paths import release_dir # ruff: ignore[import-private-name] +from timenet.sources import InstalledConnectors from timenet.types.references import DatasetRef, is_latest, split_ref from timenet_connectors.builder.backend import ConnectorBuilder from timenet_connectors.discovery import known_ids @@ -160,7 +161,7 @@ def build( ), ) except TimeNetMissingParentError as error: - error.hint = "Fetch it with --fetch-from or build it with --build-parents." + error.hint = _missing_hint(fetch_from=fetch_from, build_parents=build_parents) raise except TimeNetFileError as error: error.hint = ( @@ -175,6 +176,27 @@ def build( typer.echo(result) +def _missing_hint(*, fetch_from: str | None, build_parents: bool) -> str | None: + """Name the options of ``build`` that can get a missing release and that the command did not use. + + Args: + fetch_from: The value of ``--fetch-from``, or ``None``. + build_parents: Whether ``--build-parents`` is set. + + Returns: + The hint as a full sentence, or ``None`` if the command used both options. + """ + ways = [] + if fetch_from is None: + ways.append("fetch it with --fetch-from ") + if not build_parents: + ways.append("build it with --build-parents") + if not ways: + return None + hint = " or ".join(ways) + return f"{hint[0].upper()}{hint[1:]}." + + def _build_to( # ruff: ignore[too-many-arguments] dataset_id: str, target: WritableRegistry | Path, @@ -221,15 +243,18 @@ def _build_to( # ruff: ignore[too-many-arguments] fetch_from = fetch_from if fetch_from is not None else remote # A remote registry holds only open releases. So for a remote output, the build refuses another # release before any download, as a build into that registry would. - plan = build_release( + plan = plan_release_build( dataset_id, store=store, - fetch_from=fetch_from, - build_parents=build_parents, + # Maps --fetch-from and --build-parents onto a list of sources for the coordinator. + parents_from=[ + *([fetch_from] if fetch_from is not None else []), + *([InstalledConnectors()] if build_parents else []), + ], force=force, open_access_only=remote is not None, - options=options, ) + run_plan(plan, store, options=options) if remote is not None: copy_release(release, store, remote, progress=options.progress) return str(release.version) diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index 82b3b35ee..e61241255 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -8,22 +8,21 @@ from timenet import builders from timenet.connectors import BaseConnector -from timenet.coordinator.build import build_release -from timenet.coordinator.plan import Step, plan_parents +from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.plan import BuildStep, CopyStep, Step from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset from timenet.engine import open_parents, run_pipeline from timenet.errors import ( TimeFFormatError, TimeFValidationError, TimeNetAccessError, - TimeNetDatasetNotFoundError, TimeNetFileError, - TimeNetInvalidCardError, TimeNetMissingParentError, ) from timenet.reader import TimeFReader from timenet.registry import LocalRegistry, WritableRegistry from timenet.registry._paths import release_dir +from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset from timenet.types import Access, TimeSeriesSpec, ureg from timenet.types.access import AccessRequirement @@ -44,6 +43,17 @@ runner = CliRunner() +def _parent_steps(dataset_id: str, store: LocalRegistry, **options) -> list[Step]: + """Plan the build of ``dataset_id`` and return each step but its own build.""" + steps = plan_release_build(dataset_id, store=store, **options).steps + return [step for step in steps if not (step.action == "build" and step.release.dataset_id == dataset_id)] + + +def _build(release: DatasetRef, **fields) -> BuildStep: + """Return the build step of ``release``, with the builder that is installed for it.""" + return BuildStep(release, builders.declared(release.dataset_id), **fields) + + class _CardBackend: """A backend of another package, not ConnectorBuilder: it serves cards by dataset ID, and the plan never builds through it.""" @@ -98,20 +108,34 @@ def _install_cards(monkeypatch, **parents: dict[str, DatasetRef]) -> dict: @pytest.mark.parametrize( ("grand_in", "expected"), [ - ("output", [Step(PARENT, "build")]), - ("origin", [Step(GRAND, "copy"), Step(PARENT, "build", depends_on=(GRAND,))]), - ("nowhere", [Step(GRAND, "build"), Step(PARENT, "build", depends_on=(GRAND,))]), + ("output", lambda origin: [_build(PARENT)]), + ( + "origin", + lambda origin: [ + CopyStep(GRAND, origin), + _build(PARENT, depends_on=(GRAND,)), + ], + ), + ( + "nowhere", + lambda origin: [ + _build(GRAND), + _build(PARENT, depends_on=(GRAND,)), + ], + ), ], ) -def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_path, monkeypatch, grand_in, expected): +def test_plan_copies_or_builds_each_parent_once_before_its_children(tmp_path, monkeypatch, grand_in, expected): _install_cards(monkeypatch, child={"parent": PARENT, "grand": GRAND}, parent={"grand": GRAND}) registries = {name: LocalRegistry(tmp_path / name) for name in ("output", "origin")} if grand_in in registries: registries[grand_in].put_dataset(make_dataset()) - steps = plan_parents(CHILD.dataset_id, registries["output"], fetch_from=registries["origin"], build_parents=True) + steps = _parent_steps( + CHILD.dataset_id, registries["output"], parents_from=[registries["origin"], InstalledConnectors()] + ) - assert steps == expected + assert steps == expected(registries["origin"]) @pytest.mark.parametrize( @@ -123,20 +147,14 @@ def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_pa TimeNetMissingParentError, r"^parent release test/parent@1\.0\.0 of test/child@1\.0\.0 is not in local registry", ), - ( - {"child": {"parent": DatasetRef.parse("test/parent@2.0.0")}, "parent": {}}, - {"build_parents": True}, - TimeFValidationError, - "not that exact version", - ), ( {"child": {"parent": PARENT}, "parent": {"child": CHILD}}, - {"build_parents": True}, + {"parents_from": [InstalledConnectors()]}, TimeFValidationError, "cycle: test/child@1.0.0 -> test/parent@1.0.0 -> test/child@1.0.0", ), ], - ids=["missing", "other-version", "cycle"], + ids=["missing", "cycle"], ) # ruff: ignore[too-many-arguments, too-many-positional-arguments] def test_plan_rejects_unobtainable_parents_before_anything_runs(tmp_path, monkeypatch, cards, options, error, match): @@ -144,7 +162,7 @@ def test_plan_rejects_unobtainable_parents_before_anything_runs(tmp_path, monkey output = LocalRegistry(tmp_path / "output") with pytest.raises(error, match=match): - plan_parents(CHILD.dataset_id, output, **options) + _parent_steps(CHILD.dataset_id, output, **options) _SPEC = TimeSeriesSpec(spec_type="sine", name="Sine", unit_value=ureg.dimensionless) @@ -191,21 +209,15 @@ def test_a_declared_parent_needs_only_its_direct_payload_sources(tmp_path, monke release_dir(origin.root, SOURCE).parent.mkdir(parents=True) shutil.move(release_dir(output.root, SOURCE), release_dir(origin.root, SOURCE)) - steps = plan_parents(CHILD.dataset_id, output, fetch_from=origin) - - # DECLARED opens with SOURCE, its direct payload source, which a copy of DECLARED brings. So the - # plan does not name GRAND, SOURCE's own source. - assert steps == ([] if where == "output" else [Step(DECLARED, "copy")]) - if where == "source-in-origin": - # The output holds DECLARED, so the run copies SOURCE with its own payload sources. That - # includes GRAND, which no registry holds, so the copy fails before it copies anything. - with pytest.raises( - TimeNetMissingParentError, match=r"^payload source timenet/hello-world@1\.0\.0 of test/source" - ): - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - assert child_builds == [] - return - build_release(CHILD.dataset_id, store=output, fetch_from=origin) + # DECLARED opens with SOURCE, its direct payload source. So the plan needs neither GRAND, which + # SOURCE reads and no registry holds, nor a step for SOURCE where the output holds it. + steps = { + "output": [], + "origin": [CopyStep(DECLARED, origin)], + "source-in-origin": [CopyStep(SOURCE, origin, with_payload_sources=False)], + } + assert _parent_steps(CHILD.dataset_id, output, parents_from=[origin]) == steps[where] + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]), output) for reference in (DECLARED, SOURCE): assert output.exists(reference.dataset_id, str(reference.version)) assert not output.exists(GRAND.dataset_id, str(GRAND.version)) @@ -221,9 +233,9 @@ def test_a_payload_source_that_is_later_a_declared_parent_brings_its_own_payload _declared_without_its_sources_source(output) pins = output.get_manifest(SOURCE.dataset_id).payload_sources[SourceKey(str(GRAND))] - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build", pinned_by={SOURCE: pins})] + assert steps == [_build(GRAND, pinned_by={SOURCE: pins})] def test_the_plan_reads_each_card_from_the_backend_that_claims_the_dataset(tmp_path, monkeypatch): @@ -233,26 +245,19 @@ def test_the_plan_reads_each_card_from_the_backend_that_claims_the_dataset(tmp_p # which only the second backend knows. install_backends(monkeypatch, ConnectorBuilder(), _CardBackend(cards)) - steps = plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path), build_parents=True) + steps = _parent_steps(CHILD.dataset_id, LocalRegistry(tmp_path), parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build"), Step(PARENT, "build", depends_on=(GRAND,))] + assert steps == [ + _build(GRAND), + _build(PARENT, depends_on=(GRAND,)), + ] cards[PARENT.dataset_id] = cards[PARENT.dataset_id].model_copy(update={"dataset_version": "2.0.0"}) - with pytest.raises(TimeFValidationError, match=r"test/parent@1\.0\.0 is missing .* builds version 2\.0\.0"): - plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path), build_parents=True) - - -def test_a_parent_whose_card_cannot_be_read_fails_the_plan_with_the_card_error(tmp_path, monkeypatch): - (tmp_path / "dataset.yaml").write_text("dataset_id: [unclosed\n") - # The in-repo backend claims PARENT, whose card cannot be read. The second backend serves CHILD. - monkeypatch.setattr(backend_module, "known_ids", lambda: [PARENT.dataset_id]) - monkeypatch.setattr(backend_module, "connector_dir", lambda dataset_id: tmp_path) - install_backends(monkeypatch, ConnectorBuilder(), _CardBackend(_cards(child={"parent": PARENT}))) - - with pytest.raises(TimeNetInvalidCardError, match="could not read dataset card") as caught: - plan_parents(CHILD.dataset_id, LocalRegistry(tmp_path / "output"), build_parents=True) - - # The plan does not report a card that cannot be read as a missing connector. - assert not isinstance(caught.value, (LookupError, TimeNetDatasetNotFoundError)) + with pytest.raises( + TimeNetMissingParentError, + match=r"^parent release test/parent@1\.0\.0 of test/child@1\.0\.0 is not in local registry .* or " + r"installed connectors \(test/parent builds 2\.0\.0\)\.$", + ): + _parent_steps(CHILD.dataset_id, LocalRegistry(tmp_path), parents_from=[InstalledConnectors()]) class _Grand(BaseConnector[str]): @@ -275,18 +280,6 @@ def _parent_without_its_source(output: LocalRegistry) -> None: shutil.rmtree(release_dir(output.root, GRAND)) -def test_a_missing_payload_source_names_the_release_that_reads_it(tmp_path, monkeypatch): - _install_cards(monkeypatch, child={"parent": PARENT}, parent={"parent": GRAND}) - output = LocalRegistry(tmp_path / "output") - _parent_without_its_source(output) - - with pytest.raises(TimeNetMissingParentError) as caught: - plan_parents(CHILD.dataset_id, output) - - assert (caught.value.missing, caught.value.needed_by, caught.value.payload_source) == (GRAND, PARENT, True) - assert str(caught.value).startswith("payload source timenet/hello-world@1.0.0 of test/parent@1.0.0 is not in") - - @pytest.mark.parametrize("backend", [ValuesBackend.PARQUET, ValuesBackend.ZARR]) def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches(tmp_path, monkeypatch, backend): _install_cards(monkeypatch, child={"parent": PARENT}, parent={"parent": GRAND}) @@ -295,9 +288,9 @@ def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches( _parent_without_its_source(output) pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(GRAND))] - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) - assert steps == [Step(GRAND, "build", pinned_by={PARENT: pins})] + assert steps == [_build(GRAND, pinned_by={PARENT: pins})] built = [] def child(release, store_root, *, pinned_by, **_): @@ -311,13 +304,13 @@ def child(release, store_root, *, pinned_by, **_): monkeypatch.setattr(backend_module, "run_isolated_build", child) if backend == ValuesBackend.PARQUET: - build_release(CHILD.dataset_id, store=output, build_parents=True) + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]), output) assert output.exists(GRAND.dataset_id, str(GRAND.version)) assert built == [GRAND, CHILD] return # A Zarr rebuild writes other files than the Parquet files PARENT reads, so nothing is published. with pytest.raises(TimeFFormatError, match=r"Nothing was published\. Fetch the exact release .* or rebuild"): - build_release(CHILD.dataset_id, store=output, build_parents=True) + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]), output) assert built == [GRAND] assert not output.exists(GRAND.dataset_id, str(GRAND.version)) assert not list(release_dir(output.root, GRAND).parent.glob("*")) @@ -337,9 +330,46 @@ def test_a_payload_source_that_two_releases_read_is_built_against_the_pins_of_bo for reader in (PARENT, sibling) } - steps = plan_parents(CHILD.dataset_id, output, build_parents=True) + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) + + assert steps == [_build(GRAND, pinned_by=pins)] + + +def _repin(registry: LocalRegistry, reader: DatasetRef) -> None: + """Give the first file that ``reader`` pins of GRAND another checksum, as if another build of GRAND wrote it.""" + path = release_dir(registry.root, reader) / "manifest.json" + manifest = registry.get_manifest(reader.dataset_id) + pinned = manifest.payload_sources[SourceKey(str(GRAND))] + first, *rest = pinned.files + files = (first.model_copy(update={"checksum": "sha256:" + "0" * 64}), *rest) + sources = {**manifest.payload_sources, SourceKey(str(GRAND)): pinned.model_copy(update={"files": files})} + path.write_text(manifest.model_copy(update={"payload_sources": sources}).to_json()) + + +def test_two_readers_that_pin_one_file_differently_fail_the_plan_of_its_build(tmp_path, monkeypatch): + sibling = DatasetRef.parse("test/sibling@1.0.0") + _install_cards(monkeypatch, child={"parent": PARENT, "sibling": sibling}) + output = LocalRegistry(tmp_path / "output") + output.put_dataset(make_dataset()) + for reader in (PARENT, sibling): + _child_of(output, reader, GRAND, make_dataset().records[0].id) + _repin(output, sibling) + shutil.rmtree(release_dir(output.root, GRAND)) - assert steps == [Step(GRAND, "build", pinned_by=pins)] + # No build of GRAND can match both readers, so the plan fails before any step runs. + with pytest.raises(TimeFFormatError, match=f"pin .* of {GRAND} as different files, so no build"): + plan_release_build(CHILD.dataset_id, store=output, parents_from=[InstalledConnectors()]) + + +def test_a_held_payload_source_that_breaks_a_pin_fails_the_plan(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"parent": PARENT}) + output = LocalRegistry(tmp_path / "output") + output.put_dataset(make_dataset()) + _child_of(output, PARENT, GRAND, make_dataset().records[0].id) + _repin(output, PARENT) + + with pytest.raises(TimeFFormatError, match=f"payload source {GRAND} differs from the release {PARENT}"): + plan_release_build(CHILD.dataset_id, store=output) def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): @@ -349,20 +379,23 @@ def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(GRAND))] with pytest.raises(TimeNetMissingParentError, match=r"^payload source timenet/hello-world@1\.0\.0 of test/parent"): - plan_parents(PARENT.dataset_id, output) + _parent_steps(PARENT.dataset_id, output) + connectors = [InstalledConnectors()] # The source that PARENT reads builds against its pins. - assert plan_parents(PARENT.dataset_id, output, build_parents=True) == [ - Step(GRAND, "build", pinned_by={PARENT: pins}) + assert _parent_steps(PARENT.dataset_id, output, parents_from=connectors) == [ + _build(GRAND, pinned_by={PARENT: pins}) ] # A rebuild of PARENT needs its declared parents instead, and pins the new files. - assert plan_parents(PARENT.dataset_id, output, build_parents=True, force_build=True) == [Step(GRAND, "build")] + assert _parent_steps(PARENT.dataset_id, output, parents_from=connectors, force=True) == [_build(GRAND)] # A registry that holds the source fills the gap: the local PARENT is reused and GRAND is copied. origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) - assert plan_parents(PARENT.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] + assert _parent_steps(PARENT.dataset_id, output, parents_from=[origin]) == [ + CopyStep(GRAND, origin, with_payload_sources=False) + ] -def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkeypatch, child_builds): +def test_a_damaged_parent_is_copied_again_but_never_built_over(tmp_path, monkeypatch, child_builds): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") parent = make_dataset() @@ -372,12 +405,14 @@ def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkey (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") with pytest.raises(TimeNetFileError, match=r"'control\.duckdb' has 9 bytes, expected [\d,]+ bytes\.$") as caught: - plan_parents(CHILD.dataset_id, output, build_parents=True) + _parent_steps(CHILD.dataset_id, output, parents_from=[InstalledConnectors()]) assert caught.value.release == PARENT - assert plan_parents(CHILD.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - assert plan_parents(CHILD.dataset_id, output) == [] + assert _parent_steps(CHILD.dataset_id, output, parents_from=[origin, InstalledConnectors()]) == [ + CopyStep(PARENT, origin) + ] + run_plan(plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]), output) + assert _parent_steps(CHILD.dataset_id, output) == [] assert child_builds == [(CHILD.dataset_id, False)] @@ -413,21 +448,17 @@ def put_release(self, from_directory): _GATED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} -@pytest.mark.parametrize("gated", ["child-card", "held-parent", "parent-card"]) -def test_open_access_only_refuses_a_plan_that_visits_a_release_that_is_not_open(tmp_path, monkeypatch, gated): +def test_open_access_only_refuses_a_parent_to_build_whose_card_is_not_open(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) - gated_ref = CHILD if gated == "child-card" else PARENT - cards[gated_ref.dataset_id] = cards[gated_ref.dataset_id].model_copy(update=_GATED) + cards[PARENT.dataset_id] = cards[PARENT.dataset_id].model_copy(update=_GATED) store = LocalRegistry(tmp_path / "store") - if gated == "held-parent": - store.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) + _parent_steps(CHILD.dataset_id, store, parents_from=[InstalledConnectors()], open_access_only=True) error = caught.value - requirement = AccessRequirement(gated_ref, Access.CREDENTIALED, _GATED["access_url"]) - assert (error.dataset, error.access, error.requirements) == (gated_ref, Access.CREDENTIALED, (requirement,)) + requirement = AccessRequirement(PARENT, Access.CREDENTIALED, _GATED["access_url"]) + assert (error.dataset, error.access, error.requirements) == (PARENT, Access.CREDENTIALED, (requirement,)) def test_open_access_only_names_the_access_page_of_a_held_parent_from_its_lineage(tmp_path, monkeypatch): @@ -440,7 +471,7 @@ def test_open_access_only_names_the_access_page_of_a_held_parent_from_its_lineag store.put_dataset(held) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) + _parent_steps(CHILD.dataset_id, store, parents_from=[InstalledConnectors()], open_access_only=True) assert (caught.value.dataset, caught.value.access) == (PARENT, Access.CREDENTIALED) assert caught.value.requirements == (AccessRequirement(GRAND, Access.CREDENTIALED, _GATED["access_url"]),) @@ -554,39 +585,6 @@ def test_load_copies_a_missing_payload_source_through_fetch_from(tmp_path, monke assert len(dataset.records) == 2 -class _NoDownloads(_Hosted): - """A registry that is not local and fails the test if a file is read from it.""" - - def open_file(self, dataset_id, version, relpath): - pytest.fail(f"downloaded {dataset_id}@{version}/{relpath}") - - -@pytest.mark.parametrize("page", ["own", "ancestor"]) -def test_a_fetch_of_a_release_that_is_not_open_fails_before_any_download(tmp_path, monkeypatch, page): - cards = _install_cards(monkeypatch, child={"grand": GRAND, "parent": PARENT}, parent={}) - origin = _NoDownloads(tmp_path / "origin") - # GRAND is open and would be fetched first. - origin.put_dataset(make_dataset()) - ancestor = DatasetRef.parse("test/ancestor@1.0.0") - if page == "own": - parent, requirement = TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED)), PARENT - else: - # The card of PARENT stays open. Its lineage holds an ancestor, which is credentialed. - parent = TimeFDataset(metadata=cards[PARENT.dataset_id]) - card = make_dataset().metadata.model_copy(update={"dataset_id": ancestor.dataset_id, **_GATED}) - parent._import_state.lineage[str(ancestor)] = ParentDatasetCard(metadata=card) - requirement = ancestor - origin.put_dataset(parent) - output = LocalRegistry(tmp_path / "out") - - with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD.dataset_id, store=output, fetch_from=origin) - - assert caught.value.dataset == PARENT - assert caught.value.requirements == (AccessRequirement(requirement, Access.CREDENTIALED, _GATED["access_url"]),) - assert not output.exists(GRAND.dataset_id, str(GRAND.version)) - - def test_a_pinned_parent_build_hands_its_pins_to_the_isolated_child(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, parent={"parent": GRAND}) output = LocalRegistry(tmp_path / "output") @@ -762,3 +760,129 @@ def test_force_is_a_usage_error_if_the_remote_out_holds_the_version(tmp_path, mo assert "a published version never changes" in " ".join(output.replace("│", " ").split()) assert calls == [(CHILD.dataset_id, False)] assert _files(local.root) == held + + +def test_the_plan_drops_the_store_from_parents_from(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + store = LocalRegistry(tmp_path / "store") + + # The list holds the store itself and another registry over its directory. The plan drops both, + # so the error names the store once. + with pytest.raises(TimeNetMissingParentError) as missing: + _parent_steps(CHILD.dataset_id, store, parents_from=[store, LocalRegistry(store.root)]) + + assert missing.value.searched == str(store) + + +def test_a_parent_that_a_registry_lists_without_serving_is_built_by_the_connectors(tmp_path, monkeypatch): + cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + hosted = _Hosted(tmp_path / "hosted") + hosted.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED))) + output = LocalRegistry(tmp_path / "out") + + steps = _parent_steps(CHILD.dataset_id, output, parents_from=[hosted, InstalledConnectors()]) + + assert steps == [_build(PARENT)] + # Without the connectors, the plan names the access page of the release. + with pytest.raises(TimeNetAccessError) as caught: + _parent_steps(CHILD.dataset_id, output, parents_from=[hosted]) + assert caught.value.requirements == (AccessRequirement(PARENT, Access.CREDENTIALED, _GATED["access_url"]),) + + +@pytest.mark.parametrize("answer", ["not-found", "not-served"]) +def test_a_damaged_copy_in_the_output_stops_the_plan_if_no_registry_serves_the_release(tmp_path, monkeypatch, answer): + cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) + output, hosted = LocalRegistry(tmp_path / "output"), _Hosted(tmp_path / "hosted") + parent = make_dataset() + parent._metadata = cards[PARENT.dataset_id] + output.put_dataset(parent) + (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") + if answer == "not-served": + hosted.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED))) + + # The damage, not the missing release or its access page, is what the caller must fix. + with pytest.raises(TimeNetFileError) as caught: + _parent_steps(CHILD.dataset_id, output, parents_from=[hosted]) + + assert caught.value.release == PARENT + + +def test_the_requested_build_waits_for_a_payload_source_of_a_held_parent(tmp_path, monkeypatch): + _install_cards(monkeypatch, child={"declared": DECLARED}) + output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + _child_of(origin, SOURCE, GRAND, make_dataset().records[0].id) + _child_of(origin, DECLARED, SOURCE, 2) + # The output holds DECLARED, but not SOURCE, whose files DECLARED reads. + shutil.copytree(release_dir(origin.root, DECLARED), release_dir(output.root, DECLARED)) + + plan = plan_release_build(CHILD.dataset_id, store=output, parents_from=[origin]) + + # DECLARED has no step, so the build of CHILD waits for the step of SOURCE behind it. + assert plan.steps == ( + CopyStep(SOURCE, origin, with_payload_sources=False), + _build(CHILD, depends_on=(SOURCE,)), + ) + + +@pytest.mark.parametrize( + ("entry", "hint"), [("cli", "Build it with --build-parents."), ("api", "Copy it with fetch_from=.")] +) +def test_the_hint_names_only_the_option_that_the_build_did_not_use(tmp_path, monkeypatch, entry, hint): + # No registry holds the parent, and no installed connector builds it. The CLI build uses + # --fetch-from, and the API build uses build_parents=True. + cards = _install_cards(monkeypatch, child={"parent": PARENT}) + monkeypatch.setattr(cli_module, "known_ids", lambda: list(cards)) + monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) + out = tmp_path / "out" + + if entry == "cli": + options = ["--fetch-from", str(tmp_path / "origin")] + error = runner.invoke(app, ["build", CHILD.dataset_id, "--out", str(out), *options]).exception + else: + with pytest.raises(TimeNetMissingParentError) as caught: + timenet_connectors.build(CHILD.dataset_id, out=out, build_parents=True) + error = caught.value + + assert isinstance(error, TimeNetMissingParentError) + assert error.hint == hint + + +def _count_reads(monkeypatch, registry) -> dict[str, int]: + """Count each manifest that is read from a registry, by release.""" + reads, get_manifest = {}, registry.get_manifest + + def counted(dataset_id, version=None): + reads[f"{dataset_id}@{version}"] = reads.get(f"{dataset_id}@{version}", 0) + 1 + return get_manifest(dataset_id, version) + + monkeypatch.setattr(registry, "get_manifest", counted) + return reads + + +def test_a_plan_reads_each_manifest_once_from_each_registry(tmp_path, monkeypatch, child_builds): + # SOURCE is a declared parent and a payload source of DECLARED, so the walk meets it twice. The + # registry answers for SOURCE first as the payload source. + parents = {"declared": DECLARED, "source": SOURCE} + _install_cards(monkeypatch, child=parents) + layout = LocalRegistry(tmp_path / "hosted") + layout.put_dataset(make_dataset()) + _child_of(layout, SOURCE, GRAND, make_dataset().records[0].id) + _child_of(layout, DECLARED, SOURCE, 2) + hosted, output = _Hosted(layout.root), LocalRegistry(tmp_path / "output") + hosted_reads, output_reads = _count_reads(monkeypatch, hosted), _count_reads(monkeypatch, output) + + plan = plan_release_build(CHILD.dataset_id, store=output, parents_from=[hosted]) + + assert plan.steps[:-1] == tuple(CopyStep(reference, hosted) for reference in parents.values()) + # The registry that is not local answers through its manifest, and the plan reads that manifest + # only once. + assert hosted_reads == {str(reference): 1 for reference in (DECLARED, SOURCE, GRAND)} + # The output is read for the requested release too, which it does not hold. + assert output_reads == {str(reference): 1 for reference in (CHILD, DECLARED, SOURCE)} + hosted_reads.clear() + run_plan(plan, output) + # Each copy reads the manifests of its release and of the payload sources that it brings once + # more, from the registry as it is when the copy runs. SOURCE is in both copies. + assert hosted_reads == {str(DECLARED): 1, str(SOURCE): 2, str(GRAND): 1} + assert child_builds == [(CHILD.dataset_id, False)] diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index 6f4234227..d39839f25 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -16,7 +16,7 @@ from timenet.builders import installed_builder from timenet.config import settings -from timenet.coordinator.build import build_release +from timenet.coordinator.build import plan_release_build, run_plan from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, @@ -32,6 +32,7 @@ from timenet.registry.base import ProgressCallback from timenet.registry.dependencies import prepare_local_copy from timenet.registry.version import DatasetVersion +from timenet.sources import InstalledConnectors from timenet.types import DatasetMetadata, Domain, License, Task from timenet.types.access import require_open_access from timenet.types.references import DatasetRef, is_latest, split_ref @@ -405,7 +406,7 @@ def _prepare( Raises: TimeNetMissingParentError: If copying and building cannot provide a missing payload source. TimeNetInvalidCardError: If reading a connector's card fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and build_release + """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and the build try: manifest = registry.get_manifest(dataset_id, version) except TimeNetDatasetNotFoundError as miss: @@ -425,11 +426,15 @@ def _prepare( if found is None or found.release != source: raise pins = next(pinned for pinned in manifest.payload_sources.values() if pinned.release == source) - # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the - # coordinator until the client takes a list of sources. - build_release( - str(source), store=registry, fetch_from=fetch_from, build_parents=True, pinned_by={reference: pins} + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto a list + # of sources until the client takes the list itself. + plan = plan_release_build( + str(source), + store=registry, + parents_from=[*([fetch_from] if fetch_from is not None else []), InstalledConnectors()], + pinned_by={reference: pins}, ) + run_plan(plan, registry) built.add(source) @@ -453,7 +458,7 @@ def _obtain( # ruff: ignore[too-many-arguments] TimeNetDatasetNotFoundError: If neither ``fetch_from`` nor a connector provides the version. TimeNetVersionUnavailableError: If the connector declares a different version. TimeNetInvalidCardError: If reading the connector's card fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by build_release + """ # ruff: ignore[docstring-extraneous-exception] - raised by the build if fetch_from is not None: try: return fetch_from.get_manifest(dataset_id, version) @@ -464,14 +469,14 @@ def _obtain( # ruff: ignore[too-many-arguments] if hint is None: raise miss raise TimeNetDatasetNotFoundError(f"{miss}. {hint}") from miss - # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the coordinator - # until the client takes a list of sources. The build prepares the declared parents of the - # version in the same order: reuse, copy, then build. Sentinels such as "latest" are not pins, - # so a fresh build satisfies them. - build_release( + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto a list of + # sources until the client takes the list itself. The build prepares the declared parents of + # the version in the same order: reuse, copy, then build. Sentinels such as "latest" are not + # pins, so a fresh build satisfies them. + plan = plan_release_build( dataset_id if is_latest(version) else f"{dataset_id}@{version}", store=registry, - fetch_from=fetch_from, - build_parents=True, + parents_from=[*([fetch_from] if fetch_from is not None else []), InstalledConnectors()], ) + run_plan(plan, registry) return registry.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 1fab16f92..9aa22fdd5 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -1,146 +1,106 @@ -"""Prepare parent releases and build a dataset into a local registry.""" +"""Plan and run release builds with dependencies first.""" -from collections.abc import Mapping +from collections.abc import Mapping, Sequence +from typing import assert_never from timenet.builders import BuildOptions, declared -from timenet.coordinator.plan import Plan, Step, plan_parents -from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError -from timenet.manifest import Manifest, PayloadSource -from timenet.progress import ProgressCallback, StepFinished, StepStarted +from timenet.coordinator.plan import BuildStep, CopyStep, Plan, plan_steps +from timenet.manifest import PayloadSource +from timenet.progress import StepFinished, StepStarted from timenet.registry import copy_release -from timenet.registry._paths import release_dir -from timenet.registry.base import BaseRegistry from timenet.registry.dependencies import prepare_local_copy from timenet.registry.local import LocalRegistry -from timenet.registry.shared import complete_copy +from timenet.sources import ReleaseSource from timenet.types.references import DatasetRef, is_latest, split_ref -def build_release( # ruff: ignore[too-many-arguments] - dataset: str, +def plan_release_build( # ruff: ignore[too-many-arguments] + dataset_id: str, *, store: LocalRegistry, - fetch_from: BaseRegistry | None = None, - build_parents: bool = False, + parents_from: Sequence[ReleaseSource] = (), force: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, open_access_only: bool = False, - options: BuildOptions | None = None, ) -> Plan: - """Prepare a release's dependencies and build it into a local registry. + """Plan a release build in a local registry. - Copy missing dependencies from ``fetch_from``, or build them if ``build_parents`` is set. - Reuse an existing release unless ``force`` is set. + The plan reuses an intact release unless ``force`` is true. + Planning leaves the registry unchanged. Args: - dataset: A dataset ID such as ``org/name``, or an exact release such as ``org/name@1.2.0``. - store: The local registry to build into. - fetch_from: The registry to copy missing parents from, or ``None`` to copy nothing. - build_parents: Build missing parents with installed connectors. - force: Rebuild the requested release and prepare its parents even if the release exists. - pinned_by: Releases that already read the requested release's files, with the pins the new build must match. - open_access_only: Require ``access: open`` for all releases and their lineage before running the plan. - options: Build settings, or ``None`` for defaults. + dataset_id: A dataset ID such as ``org/name``, or an exact release such as ``org/name@1.2.0``. + 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: - The completed plan, with steps in execution order. + A plan with steps in execution order. Raises: - TimeNetDatasetNotFoundError: If no installed backend builds the dataset or a missing parent. - TimeNetVersionUnavailableError: If the connector declares a different version. - TimeNetVersionConflictError: If ``store`` holds a different copy of a release being copied. - TimeFValidationError: If the dataset reference, backend, or dependency plan is invalid. - TimeNetMissingParentError: If a parent is missing and ``build_parents`` is not set. - TimeNetAccessError: If a release or its lineage lacks required open access. - TimeNetInvalidCardError: If an installed connector's card cannot be read. - TimeNetFileError: If ``store`` holds a damaged parent and no replacement is available. - TimeFFormatError: If a build omits or changes a pinned file. - TimeNetBuildError: If a build fails. - """ # ruff: ignore[docstring-extraneous-exception] - raised by the plan, the copy and the backends - options = options or BuildOptions() - dataset_id, version = split_ref(dataset) - # 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, + TimeNetDatasetNotFoundError: If no installed connector builds the dataset. + TimeNetVersionUnavailableError: If the requested version differs from the connector's card. + TimeNetMissingParentError: If no source provides a parent or a payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by declared and the plan + dataset_id, version = split_ref(dataset_id) + _, steps = plan_steps( + # A connector builds only the version of its card, so "latest" asks for that version. + declared(dataset_id, None if is_latest(version) else version), store, - fetch_from=fetch_from, - build_parents=build_parents, + parents_from=parents_from, force_build=force, open_access_only=open_access_only, + pinned_by=pinned_by, ) - if force or not store.exists(requested.dataset_id, str(requested.version)): - planned = {step.release for step in steps} - depends_on = tuple(parent for parent in card.parents.values() if parent in planned) - steps.append(Step(requested, "build", depends_on=depends_on, pinned_by=pinned_by or {})) - if fetch_from is None and any(step.action == "copy" for step in steps): - raise TimeFValidationError(f"the plan of {requested} copies a release, but has no registry to copy from") - for step in steps: - if step.action == "build": - _build(step, store, force=force and step.release == requested, options=options) - elif fetch_from is not None: - _copy(step.release, fetch_from, store, progress=options.progress) return Plan(tuple(steps)) -def _copy( - release: DatasetRef, fetch_from: BaseRegistry, store: LocalRegistry, *, progress: ProgressCallback | None -) -> None: - """Copy a release and its required payload sources into the local registry. - - If the release is intact, copy only its missing or damaged payload sources. - - Args: - release: The release to copy. - fetch_from: The registry to copy from. - store: The destination registry. - progress: The callback for copy progress events, or ``None``. - """ - held = _intact_manifest(store, release) - if held is None: - copy_release(release, fetch_from, store, progress=progress) - return - for pinned in held.payload_sources.values(): - if _intact_manifest(store, pinned.release) is None: - copy_release(pinned.release, fetch_from, store, progress=progress) - # The copied payload sources must hold the files that the release pins. - prepare_local_copy(store.root, release) - - -def _intact_manifest(store: LocalRegistry, reference: DatasetRef) -> Manifest | None: - """Read a release's manifest and check its file sizes. +def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None: + """Run a plan's copy and build steps in order. Args: - store: The local registry. - reference: The release to check. + plan: The steps to run in dependency order. + store: The local registry to copy and build into. + options: Build settings, including the progress callback for copies and builds. - Returns: - The manifest, or ``None`` if the release or a file is missing or a file has the wrong size. - """ - try: - manifest = store.get_manifest(reference.dataset_id, str(reference.version)) - except TimeNetDatasetNotFoundError: - return None - return manifest if complete_copy(release_dir(store.root, reference), manifest) else None - - -def _build(step: Step, store: LocalRegistry, *, force: bool, options: BuildOptions) -> None: - """Build a planned release and check its required files. + Raises: + TimeNetVersionConflictError: If ``store`` holds different files for a copied release. + TimeFFormatError: If a build fails to reproduce pinned files. + TimeNetBuildError: If a build fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the copy and the backends + options = options or BuildOptions() + for step in plan.steps: + match step: + case CopyStep(): + copy_release( + step.release, + step.registry, + store, + with_payload_sources=step.with_payload_sources, + progress=options.progress, + ) + case BuildStep(): + _build(step, store, options=options) + case _: + assert_never(step) + + +def _build(step: BuildStep, store: LocalRegistry, *, options: BuildOptions) -> None: + """Build an exact release with the builder of its step and check its required files. Args: - step: The build step. + step: The build step, with ``force`` set to rebuild an existing release. store: The local registry to build into. - force: Rebuild even if ``store`` already holds the release. options: The build settings. Raises: - TimeNetDatasetNotFoundError: If no backend builds this version or the result lacks a release or payload source. - """ # ruff: ignore[docstring-extraneous-exception] - raised by declared and prepare_local_copy - builder = declared(step.release.dataset_id, step.release.version).backend + TimeNetDatasetNotFoundError: If the result lacks the release or a payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by prepare_local_copy if options.progress is not None: options.progress(StepStarted(step.release, "build")) - builder.build(step.release, store, force=force, pinned_by=step.pinned_by, options=options) + step.builder.backend.build(step.release, store, force=step.force, pinned_by=step.pinned_by, options=options) # After a successful build, the local registry must hold the exact release with its payload sources. prepare_local_copy(store.root, step.release) if options.progress is not None: diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index cf4d0fac3..d86ac72f2 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -1,16 +1,11 @@ -"""Plan how to obtain the parent releases that a build needs, before any connector runs. +"""Plan release copies and builds with dependencies first, before any connector runs.""" -Each :class:`Step` copies or builds one release after its dependencies are ready. -""" - -from collections.abc import Mapping -from dataclasses import dataclass, field +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field, replace from graphlib import CycleError, TopologicalSorter -from typing import Literal +from typing import ClassVar, Literal, TypeAlias -import pyarrow.fs as pafs - -from timenet.builders import find_card +from timenet.builders import InstalledBuilder, find_card from timenet.errors import ( TimeFFormatError, TimeFValidationError, @@ -18,25 +13,52 @@ TimeNetFileError, TimeNetMissingParentError, ) -from timenet.manifest import Manifest, PayloadSource +from timenet.manifest import FilePart, Manifest, PayloadSource from timenet.registry.base import BaseRegistry -from timenet.registry.dependencies import check_file_sizes +from timenet.registry.dependencies import ( + CopySources, + check_pins, + find_payload_sources, + validate_manifest_reference, +) from timenet.registry.local import LocalRegistry -from timenet.registry.version import DatasetVersion -from timenet.types import DatasetMetadata +from timenet.sources import ( + FoundInConnector, + FoundInRegistry, + InstalledConnectors, + NotFound, + NotServed, + ReleaseLookup, + ReleaseSource, + resolve_release, +) from timenet.types.access import require_open_access from timenet.types.references import DatasetRef +from timenet.types.version import Version @dataclass(frozen=True) -class Step: - """A plan step that copies or builds one release. +class CopyStep: + """A planned copy of one release from a registry into the local registry.""" - A copy includes the release and its missing payload sources. - """ + release: DatasetRef + registry: BaseRegistry + """The registry to copy from.""" + depends_on: tuple[DatasetRef, ...] = () + """The releases with steps that must finish before this step runs.""" + with_payload_sources: bool = True + """Also copy the payload sources that the manifest of the release lists.""" + action: ClassVar[Literal["copy"]] = "copy" + """The kind of step, for display.""" + + +@dataclass(frozen=True) +class BuildStep: + """A planned build of one release by an installed connector into the local registry.""" release: DatasetRef - action: Literal["copy", "build"] + builder: InstalledBuilder = field(hash=False) + """The installed connector that builds the release.""" depends_on: tuple[DatasetRef, ...] = () """The releases with steps that must finish before this step runs.""" pinned_by: Mapping[DatasetRef, PayloadSource] = field(default_factory=dict, hash=False) @@ -44,6 +66,14 @@ class Step: A build must reproduce every recorded file before it publishes the release. """ + force: bool = False + """Build the release again, also if the local registry holds it.""" + action: ClassVar[Literal["build"]] = "build" + """The kind of step, for display.""" + + +Step: TypeAlias = CopyStep | BuildStep +"""A planned copy or build of one release.""" @dataclass(frozen=True) @@ -53,83 +83,267 @@ class Plan: steps: tuple[Step, ...] +Role = Literal["dataset", "payload-source"] +"""The release's role as a dataset or a source of payload files. + +A release needs exactly the releases that its manifest lists as payload sources, never the payload sources of a +payload source. So a ``"dataset"`` needs its payload sources to open, and a ``"payload-source"`` supplies only its +own files. +""" + + @dataclass(frozen=True) class _Release: - """A required release, its dependencies, and its planned action.""" + """A required release with its role, planned step and dependencies.""" - opened: bool - """True if the build opens the release with its payload sources. False if the release is only a - payload source of another release, and the build reads only its pinned files.""" - action: Literal["copy", "build"] | None - """The planned action, or ``None`` if the output already holds the release.""" + role: Role + """The release's role as a dataset or a source of payload files.""" + step: Step | None + """The copy or build step before dependencies are set, or ``None`` for an existing release.""" dependencies: tuple[DatasetRef, ...] - """The releases that must be in place before this release. A release to build depends on the - parents on its card. An opened release that the plan does not build depends on its payload - sources. Other releases depend on nothing.""" + """The parent or payload source releases that must be ready first.""" @dataclass class _Graph: - """The releases a build needs and the dependencies between them.""" + """A graph of required releases and their copy or build steps.""" output: LocalRegistry - fetch_from: BaseRegistry | None - build_parents: bool + sources: Sequence[ReleaseSource] + """Sources for missing releases in preference order, excluding the output.""" open_only: bool - """Whether each visited release and its lineage must declare ``access: open``.""" - searched: str - """The registries that the plan looks in, for the message of a missing release.""" + """True if each release that the walk visits must declare ``access: open``.""" releases: dict[DatasetRef, _Release] = field(default_factory=dict) """The nodes, in the order in which the walk found them.""" - fetched: set[DatasetRef] = field(default_factory=set) - """The payload sources that the plan copies as part of another release.""" pins: dict[DatasetRef, dict[DatasetRef, PayloadSource]] = field(default_factory=dict) """For each payload source: each release that reads its files, with its pins on them.""" + answers: dict[tuple[int, DatasetRef], ReleaseLookup] = field(default_factory=dict) + """Cached source answers by source identity and release.""" + manifests: dict[tuple[int, DatasetRef], Manifest | None] = field(default_factory=dict) + """Cached manifests keyed by registry identity and release, with ``None`` for missing releases.""" + damage: dict[DatasetRef, TimeNetFileError | None] = field(default_factory=dict) + """The first file error per checked local release, or ``None`` if all file sizes match.""" def visit(self, reference: DatasetRef, needed_by: DatasetRef, pinned: PayloadSource | None = None) -> None: - """Add a release and its required dependencies to the graph. + """Add a release and its dependencies to the graph. - Use :meth:`order` to check the completed graph for cycles. + The graph revisits an existing entry only when its role changes from payload source to dataset. Args: reference: The release to add. needed_by: The release that needs ``reference``. - pinned: File pins from ``needed_by``, or ``None`` to open ``reference`` as a parent. - """ - opened = pinned is None + pinned: File pins from ``needed_by``, or ``None`` for a parent dataset. + + Raises: + TimeNetFileError: If the output holds a damaged copy that no other registry provides. + TimeNetAccessError: If a source lists the release but no source serves it. + TimeNetMissingParentError: If no source provides the release. + """ # ruff: ignore[docstring-extraneous-exception] - the file and access errors are raised as values + role: Role = "dataset" if pinned is None else "payload-source" if pinned is not None: self.pins.setdefault(reference, {})[needed_by] = pinned found = self.releases.get(reference) # If the walk found a payload source first, and later finds it as a parent, it looks at the # release again with its payload sources. - if found is not None and (found.opened or not opened): + if found is not None and (found.role == "dataset" or role == "payload-source"): return - manifest, action = _locate(self.output, self.fetch_from, reference, open_only=self.open_only) - if manifest is None: - card = _card_to_build( - reference, - needed_by, - build_parents=self.build_parents, - open_only=self.open_only, - pinned=not opened, - searched=self.searched, - ) - self.releases[reference] = _Release(opened, "build", tuple(card.parents.values())) - # A build opens each of its parents with the payload sources of that parent. - for parent in card.parents.values(): - self.visit(parent, reference) - elif opened: - # To open a release, the build needs its direct payload sources. These are the releases - # whose files the release reads. The build does not need their own payload sources. - sources = list(manifest.payload_sources.values()) - self.releases[reference] = _Release(opened, action, tuple(source.release for source in sources)) + held, damage = self.in_output(reference) + if held is not None: + if self.open_only: + require_open_access(held.metadata, held.lineage) + sources = list(held.payload_sources.values()) if role == "dataset" else [] + self.releases[reference] = _Release(role, None, tuple(source.release for source in sources)) for source in sources: self.visit(source.release, reference, source) + return + answer = self.resolve(reference) + if isinstance(answer, FoundInRegistry) and not _is_same_registry(answer.registry, self.output): + self.copy(reference, role, answer.registry) + elif damage is not None: + # A copy from another registry repairs a damaged copy. Nothing else replaces it. + raise damage + elif isinstance(answer, FoundInConnector): + self.build(reference, role, answer.builder) + elif isinstance(answer, NotServed): + raise answer.access_error(reference.dataset_id) else: - self.releases[reference] = _Release(opened, action, ()) - # A copy of the release that reads this payload source also copies the payload source. - if action == "copy": - self.fetched.add(reference) + # No source provides the release, or only the output does. + raise TimeNetMissingParentError( + reference, needed_by, self.describe_searched_sources(reference), payload_source=role == "payload-source" + ) + + def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> None: + """Add a copy step and plan the payload sources that the copy does not bring. + + A dataset copy brings each payload source that ``registry`` serves. Each other payload source that the output + lacks gets its own step first. + + Args: + reference: The release to copy. + role: Whether the release is a dataset or a payload source. + registry: The registry to copy from. + + Raises: + TimeNetDatasetNotFoundError: If the registry no longer holds the release. + TimeFFormatError: If a payload source that the copy brings or reuses fails a file pin. + """ # ruff: ignore[docstring-extraneous-exception] - raised by find_payload_sources + manifest = self.manifest(registry, reference) + if manifest is None: + raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") + if self.open_only: + require_open_access(manifest.metadata, manifest.lineage) + sources = CopySources() + if role == "dataset": + sources = find_payload_sources( + reference, + manifest, + registry, + find=lambda source: self.listed(registry, source), + held=lambda source: self.in_output(source)[0], + ) + if self.open_only: + for found in (*sources.fetched.values(), *sources.held.values()): + require_open_access(found.metadata, found.lineage) + step = CopyStep(reference, registry, with_payload_sources=role == "dataset") + # The node holds its final dependencies before the walk visits them. So a release of the walk + # that depends on it closes a cycle and does not look it up again. + self.releases[reference] = _Release(role, step, tuple(sources.missing)) + for source, pinned in sources.missing.items(): + self.visit(source, reference, pinned) + + def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> None: + """Add a build step and plan the parents declared in its card. + + Args: + reference: The release to build. + role: Whether the release is a dataset or a payload source. + builder: The installed connector that builds the release. + """ + card = builder.card + if self.open_only: + require_open_access(card) + parents = tuple(card.parents.values()) + self.releases[reference] = _Release(role, BuildStep(reference, builder), parents) + # A build opens each of its parents with the payload sources of that parent. + for parent in parents: + self.visit(parent, reference) + + def resolve(self, reference: DatasetRef) -> ReleaseLookup: + """Find an exact release across the sources in preference order. + + Args: + reference: The exact release to find. + + Returns: + The first ``Found`` answer, otherwise the first ``NotServed`` answer or ``NotFound()``. + """ + return resolve_release(self.sources, reference.dataset_id, str(reference.version), ask=self._ask) + + def ask(self, source: ReleaseSource, reference: DatasetRef) -> ReleaseLookup: + """Get and cache a source's answer for an exact release. + + Args: + source: The source to query. + reference: The exact release to find. + + Returns: + The cached or newly fetched answer. + """ + key = (id(source), reference) + if key not in self.answers: + self.answers[key] = source.resolve(reference.dataset_id, reference.version) + return self.answers[key] + + def _ask(self, source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: + """Return the answer of a source for a release, from :meth:`ask` for an exact version.""" + # The plan asks only for exact versions, so it keeps only those answers. + if version is None: + return source.resolve(dataset_id, None) + return self.ask(source, DatasetRef(dataset_id=dataset_id, version=version)) + + def listed(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: + """Get a release's manifest if the registry holds the release, also if it does not serve it. + + Returns: + The manifest, or ``None`` if the registry does not hold the release. + """ + if isinstance(self.ask(registry, reference), NotFound): + return None + return self.manifest(registry, reference) + + def manifest(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: + """Read and cache a release's manifest. + + A manifest in a cached source answer takes precedence over a registry read. + + Returns: + The manifest, or ``None`` if the registry does not hold the release. + """ + key = (id(registry), reference) + if key not in self.manifests: + answer = self.answers.get(key) + if isinstance(answer, FoundInRegistry | NotServed) and answer.manifest is not None: + validate_manifest_reference(answer.manifest, reference) + self.manifests[key] = answer.manifest + else: + self.manifests[key] = registry.find_manifest(reference) + return self.manifests[key] + + def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFileError | None]: + """Check a release in the output for missing files or incorrect file sizes. + + Returns: + ``(manifest, None)`` if intact, ``(None, error)`` if damaged, else ``(None, None)``. + """ + manifest = self.manifest(self.output, reference) + if manifest is None: + return manifest, None + if reference not in self.damage: + self.damage[reference] = self.output.damage(reference, manifest) + damage = self.damage[reference] + return (manifest, None) if damage is None else (None, damage) + + def describe_searched_sources(self, reference: DatasetRef) -> str: + """Describe the output and sources searched for a missing release. + + Connector descriptions include their declared version when it differs from the requested version. + + Args: + reference: The missing release. + + Returns: + The source descriptions joined with ``" or "``. + """ + names = [str(self.output)] + for source in self.sources: + card = find_card(reference.dataset_id) if isinstance(source, InstalledConnectors) else None + if card is not None and card.dataset_version != reference.version: + names.append(f"{source} ({reference.dataset_id} builds {card.dataset_version})") + else: + names.append(str(source)) + return " or ".join(names) + + def check(self) -> None: + """Check each pinned release against the pins of every release that reads it. + + Raises: + TimeFFormatError: If a copied or held release fails a file pin, or two readers of a build pin one file + differently. + TimeNetDatasetNotFoundError: If a registry no longer holds a copied or held release. + """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins + for reference, readers in self.pins.items(): + step = self.releases[reference].step + if isinstance(step, BuildStep): + # One build must match every reader, so the readers must pin each file alike. + _check_agreement(reference, readers) + continue + # A copy reads the release from its registry. A release without a step is intact in the output. + registry = self.output if step is None else step.registry + manifest = self.manifest(registry, reference) + if manifest is None: + raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") + for reader, pinned in readers.items(): + check_pins(reader, pinned, manifest) def order(self) -> list[DatasetRef]: """Order the releases so that each release comes after the releases that it depends on. @@ -156,219 +370,120 @@ def order(self) -> list[DatasetRef]: path = " -> ".join(str(reference) for reference in (*cycle, cycle[0])) raise TimeFValidationError(f"dataset parent cycle: {path}") from error - def steps(self, requested: DatasetRef) -> list[Step]: - """Create copy and build steps with dependencies first. - - Each step names only dependencies that also have steps. - - Args: - requested: The requested release, which gets a copy step if needed but no build step. + def create_ordered_steps(self) -> list[Step]: + """Create copy and build steps in dependency order. Returns: - The steps in dependency order. + The ordered steps, including the requested build if needed. Raises: - TimeFValidationError: If the releases form a dependency cycle. + TimeFValidationError: If the dependencies form a cycle. """ # ruff: ignore[docstring-extraneous-exception] - order() raises the cycle error - chosen: dict[DatasetRef, Literal["copy", "build"]] = {} + nearest: dict[DatasetRef, tuple[DatasetRef, ...]] = {} + steps: list[Step] = [] for reference in self.order(): release = self.releases[reference] - if reference == requested and release.action == "build": - continue - if release.action == "build": - chosen[reference] = "build" - # A copy of an opened release also copies each payload source that the output does not - # hold. So if the plan copies a payload source only with the release that reads it, the - # plan copies that release again, even if the output holds it. - elif release.opened and ( - release.action == "copy" or any(source in self.fetched for source in release.dependencies) - ): - chosen[reference] = "copy" - return [ - Step( - reference, - action, - depends_on=tuple(need for need in self.releases[reference].dependencies if need in chosen), - pinned_by=self.pins.get(reference, {}) if action == "build" else {}, - ) - for reference, action in chosen.items() - ] + # The order puts each dependency first, so its nearest releases with a step are known. + needs: dict[DatasetRef, None] = {} + for need in release.dependencies: + needs.update(dict.fromkeys((need,) if self.releases[need].step is not None else nearest[need])) + nearest[reference] = tuple(needs) + step = release.step + if isinstance(step, BuildStep): + step = replace(step, pinned_by=self.pins.get(reference, {})) + if step is not None: + steps.append(replace(step, depends_on=nearest[reference])) + return steps + + +def _check_agreement(release: DatasetRef, readers: Mapping[DatasetRef, PayloadSource]) -> None: + """Check that the readers of a release pin each of its files alike. + Args: + release: The release whose files are pinned. + readers: The releases that pin files of ``release``, with their file requirements. -def plan_parents( # ruff: ignore[too-many-arguments] - dataset_id: str, + Raises: + TimeFFormatError: If two readers pin one file as different files. + """ + pinned: dict[str, tuple[DatasetRef, FilePart]] = {} + for reader, source in readers.items(): + for part in source.files: + first_reader, first_part = pinned.setdefault(part.path, (reader, part)) + if first_part != part: + raise TimeFFormatError( + f"{first_reader} and {reader} pin {part.path!r} of {release} as different files, " + f"so no build of {release} can match both" + ) + + +def plan_steps( # ruff: ignore[too-many-arguments] + builder: InstalledBuilder, output: LocalRegistry, *, - fetch_from: BaseRegistry | None = None, - build_parents: bool = False, - force_build: bool = False, - open_access_only: bool = False, -) -> list[Step]: - """Plan the copy and build steps for a dataset's parents. + 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 its dependencies. - Copy missing parents and their payload sources from ``fetch_from``, or build them if enabled. + A damaged copy of the requested release in the output gets a forced build. Args: - dataset_id: The dataset to build. + builder: The installed connector that builds the dataset. output: The local registry that the build writes to. - fetch_from: The registry to copy missing parents from, or ``None``. - build_parents: Build missing parents with their installed connectors. - force_build: Plan parents for a rebuild even if the requested release already exists. - open_access_only: Require ``access: open`` for each release, including the requested release. + parents_from: Sources for missing dependencies, in preference order. + force_build: Plan a rebuild even if the requested release exists. + open_access_only: Require ``access: open`` for every needed release. + pinned_by: File pins grouped by dependent release that the requested build must preserve. Returns: - The copy and build steps, with dependencies first. - - Raises: - LookupError: If no installed connector declares a card for ``dataset_id``. - TimeNetInvalidCardError: If an installed connector's card cannot be read. - TimeFValidationError: If parents form a cycle or versions conflict. - TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false. - TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it. - TimeNetFileError: If a local parent is damaged and no replacement is available. - TimeNetAccessError: If a release or its lineage lacks required open access. - """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk - card = find_card(dataset_id) - if card is None: - raise LookupError(f"no installed connector declares a card for {dataset_id!r}") + The requested release and ordered steps, with its own build last if needed. + """ + # The output is never a source for itself. + parents_from = [source for source in parents_from if not _is_same_registry(source, output)] + card = builder.card if open_access_only: require_open_access(card) - # For a dataset without parents, the plan does not read the output, so the build makes no extra request. - if not card.parents: - return [] - root = DatasetRef(dataset_id=card.dataset_id, version=card.dataset_version) + root = builder.release + 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 = None if force_build else _manifest(output, root) + held, damage = (None, None) if force_build else graph.in_output(root) if held is not None and open_access_only: # The installed card has no lineage, so the check above does not see an ancestor that is not open. require_open_access(held.metadata, held.lineage) - searched = str(output) if fetch_from is None else f"{output} or {fetch_from}" - graph = _Graph(output, fetch_from, build_parents=build_parents, open_only=open_access_only, searched=searched) # The graph holds the requested release before the walk starts. So a release that depends on it # closes a cycle, and the walk does not look it up again. if held is None: - graph.releases[root] = _Release(opened=True, action="build", dependencies=tuple(card.parents.values())) + step = BuildStep(root, builder, force=force_build or damage is not None) + graph.releases[root] = _Release(role="dataset", step=step, dependencies=tuple(card.parents.values())) for parent in card.parents.values(): graph.visit(parent, root) else: sources = list(held.payload_sources.values()) graph.releases[root] = _Release( - opened=True, action=None, dependencies=tuple(source.release for source in sources) + role="dataset", step=None, dependencies=tuple(source.release for source in sources) ) for source in sources: graph.visit(source.release, root, source) - return graph.steps(root) - - -def _manifest(registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: - """Read the manifest of one release. - - This reads no other file and resolves no payload source. - - Returns: - The manifest, or ``None`` if the registry does not hold the release. - - Raises: - TimeFFormatError: If the manifest declares another dataset ID or version than ``reference``. - """ - try: - manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) - except TimeNetDatasetNotFoundError: - return None - if manifest.dataset_id != reference.dataset_id or manifest.metadata.dataset_version != reference.version: - raise TimeFFormatError(f"manifest for {reference} declares a different dataset ID or version") - return manifest + graph.check() + return root, graph.create_ordered_steps() -def _locate( - output: LocalRegistry, fetch_from: BaseRegistry | None, reference: DatasetRef, *, open_only: bool -) -> tuple[Manifest | None, Literal["copy"] | None]: - # The damage is raised from a variable, the access error by require_open_access. - # ruff: ignore[docstring-extraneous-exception] - """Find a release in the output or the source registry. +def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool: + """Check whether a source is the output registry. - A local copy is intact when all files exist at their recorded sizes. - Enforce open access for the release and its lineage when ``open_only`` or a copy requires it. + Args: + source: The source to compare. + output: The destination registry. Returns: - The manifest and action (``None`` to reuse, ``"copy"`` to fetch), or ``(None, None)`` if absent. - - Raises: - TimeNetFileError: If the local copy is damaged and the source registry lacks the release. - TimeNetAccessError: If the release or its lineage lacks required open access. + ``True`` for the same object or local registries with the same resolved directory. """ - manifest = _manifest(output, reference) - damage = None - if manifest is not None: - directory = output.root / reference.dataset_id / str(reference.version) - try: - check_file_sizes( - DatasetVersion(manifest, pafs.LocalFileSystem(), str(directory)), - manifest.files.all_files(), - owner=f"committed release {reference} in {output}", - ) - except TimeNetFileError as error: - manifest, damage = None, error - if manifest is not None: - if open_only: - require_open_access(manifest.metadata, manifest.lineage) - return manifest, None - fetched = None if fetch_from is None else _manifest(fetch_from, reference) - if fetched is not None: - # A registry that requires open access serves only releases whose cards declare access: open. - # So the plan refuses another release before any download. - if open_only or (fetch_from is not None and fetch_from.requires_open_access): - require_open_access(fetched.metadata, fetched.lineage) - return fetched, "copy" - if damage is not None: - raise damage - return None, None - - -def _card_to_build( # ruff: ignore[too-many-arguments] - reference: DatasetRef, - needed_by: DatasetRef, - *, - build_parents: bool, - open_only: bool, - pinned: bool, - searched: str, -) -> DatasetMetadata: - """Find a connector card for building a missing release. - - Returns: - The connector card for the exact dataset ID and version in ``reference``. - - Raises: - TimeNetMissingParentError: If ``build_parents`` is false. - TimeNetAccessError: If ``open_only`` is set and the card does not declare ``access: open``. - """ # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access - if not build_parents: - raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned) - card = _exact_card(reference) - if open_only: - require_open_access(card) - return card - - -def _exact_card(reference: DatasetRef) -> DatasetMetadata: - """Return the card of the installed connector that builds exactly ``reference``. - - Returns: - The card of the connector. - - Raises: - TimeNetDatasetNotFoundError: If no installed connector builds the dataset. - TimeFValidationError: If the installed connector declares another version. - TimeNetInvalidCardError: If the card of the installed connector cannot be read. - """ # ruff: ignore[docstring-extraneous-exception] - find_card raises the card error - card = find_card(reference.dataset_id) - if card is None: - raise TimeNetDatasetNotFoundError(f"parent release {reference} is missing and no installed connector builds it") - if card.dataset_version != reference.version: - raise TimeFValidationError( - f"parent release {reference} is missing and the installed connector builds " - f"version {card.dataset_version}, not that exact version" - ) - return card + if source is output: + return True + return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve() diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index b9ec678e9..ba07a0f23 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -43,8 +43,10 @@ def copy_release( TimeNetVersionConflictError: If ``dest`` holds the release or a payload source with other files or another card. TimeNetAccessError: If a release lacks open access where the source or destination requires it. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. - TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` lacks a complete copy. - TimeFFormatError: If a manifest names the wrong release or pinned files differ. + TimeNetMissingParentError: If a payload source is neither in ``source`` nor complete in + ``dest``. The error names both registries. + TimeFFormatError: If a manifest names another release, or a payload source differs from the + files that the release pins. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload @@ -146,7 +148,7 @@ def _manifests( if listed is not None: # The source lists the payload source, but does not serve its files. require_open_access(listed.metadata, listed.lineage) - raise TimeNetMissingParentError(missing, reference, str(source), payload_source=True) + raise TimeNetMissingParentError(missing, reference, f"{source} or {dest}", payload_source=True) return {**sources.fetched, **sources.held, reference: manifest} diff --git a/packages/timenet/src/timenet/sources.py b/packages/timenet/src/timenet/sources.py index a72aaaea0..3689f8baa 100644 --- a/packages/timenet/src/timenet/sources.py +++ b/packages/timenet/src/timenet/sources.py @@ -1,6 +1,6 @@ """Find dataset releases across registries and installed connectors.""" -from collections.abc import Sequence +from collections.abc import Callable, Sequence from dataclasses import dataclass, field from typing import TYPE_CHECKING, TypeAlias, Union @@ -128,7 +128,13 @@ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: # return FoundInConnector(builder) -def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: str | None) -> ReleaseLookup: +Ask: TypeAlias = Callable[[ReleaseSource, str, Version | None], ReleaseLookup] +"""A function that asks one source for a release: ``ask(source, dataset_id, version)``.""" + + +def resolve_release( + sources: Sequence[ReleaseSource], dataset_id: str, version: str | None, *, ask: Ask | None = None +) -> ReleaseLookup: """Find a source that provides the requested dataset release. An exact version comes from the first source that serves it, and no version means the newest version of the first @@ -139,6 +145,7 @@ def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: sources: The sources, in order. dataset_id: The dataset ID. version: An exact version, ``"latest"``, or ``None`` or ``""`` for no version. + ask: The function that asks one source, or ``None`` to call the source's ``resolve()``. Returns: A ``Found`` result if available, otherwise the first ``NotServed`` result or ``NotFound``. @@ -146,16 +153,24 @@ def resolve_release(sources: Sequence[ReleaseSource], dataset_id: str, version: Raises: TimeFValidationError: If ``version`` is neither ``"latest"`` nor a ``major.minor.patch`` version. """ # ruff: ignore[docstring-extraneous-exception] - raised by Version.parse + ask = ask or _ask if version == LATEST_VERSION: - return _resolve_latest(sources, dataset_id) - return _resolve_first(sources, dataset_id, Version.parse(version) if version else None) + return _resolve_latest(sources, dataset_id, ask) + return _resolve_first(sources, dataset_id, Version.parse(version) if version else None, ask) + + +def _ask(source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: + """Return the answer of a source for a release.""" + return source.resolve(dataset_id, version) -def _resolve_first(sources: Sequence[ReleaseSource], dataset_id: str, version: Version | None) -> ReleaseLookup: +def _resolve_first( + sources: Sequence[ReleaseSource], dataset_id: str, version: Version | None, ask: Ask +) -> ReleaseLookup: """Return the answer of the first source that serves ``version``, or its newest for ``None``, else the first listing.""" listed: NotServed | None = None for source in sources: - answer = source.resolve(dataset_id, version) + answer = ask(source, dataset_id, version) if isinstance(answer, Found): return answer if isinstance(answer, NotServed) and listed is None: @@ -163,23 +178,25 @@ def _resolve_first(sources: Sequence[ReleaseSource], dataset_id: str, version: V return NotFound() if listed is None else listed -def _resolve_latest(sources: Sequence[ReleaseSource], dataset_id: str) -> ReleaseLookup: +def _resolve_latest(sources: Sequence[ReleaseSource], dataset_id: str, ask: Ask) -> ReleaseLookup: """Return the highest version that any source provides, from the first source that provides it.""" best: Found | None = None listed: list[tuple[ReleaseSource, NotServed]] = [] before_best = 0 for source in sources: - answer = source.resolve(dataset_id, None) + answer = ask(source, dataset_id, None) if isinstance(answer, NotServed): listed.append((source, answer)) elif isinstance(answer, Found) and (best is None or answer.version > best.version): best, before_best = answer, len(listed) if best is None: return listed[0][1] if listed else NotFound() - return _ask_again(listed[:before_best], dataset_id, best.version) or best + return _ask_again(listed[:before_best], dataset_id, best.version, ask) or best -def _ask_again(listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: str, version: Version) -> Found | None: +def _ask_again( + listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: str, version: Version, ask: Ask +) -> Found | None: """Ask each source that lists a higher version without serving it whether it serves ``version``. Returns: @@ -187,7 +204,7 @@ def _ask_again(listed: Sequence[tuple[ReleaseSource, NotServed]], dataset_id: st """ for source, answer in listed: if answer.version > version: - again = source.resolve(dataset_id, version) + again = ask(source, dataset_id, version) if isinstance(again, Found): return again return None diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index be5ca53cc..103af8160 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -2,16 +2,17 @@ import pytest +from timenet import builders from timenet.builders import BuildOptions -from timenet.coordinator.build import build_release -from timenet.coordinator.plan import Step +from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.plan import BuildStep, CopyStep from timenet.dataset import TimeFDataset -from timenet.errors import TimeNetAccessError, TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError +from timenet.errors import TimeNetAccessError, TimeNetVersionUnavailableError from timenet.progress import StepFinished, StepStarted from timenet.reader import TimeFReader from timenet.registry import LocalRegistry from timenet.registry._paths import release_dir -from timenet.registry.shared import complete_copy +from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset from timenet.types import Access from timenet.types.reference_keys import ParentAlias, SourceKey @@ -25,6 +26,11 @@ _GATED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} +def _build(release: DatasetRef, **fields) -> BuildStep: + """Return the build step of ``release``, with the builder that is installed for it.""" + return BuildStep(release, builders.declared(release.dataset_id), **fields) + + def _card(reference, **update): """Return the card of the demo dataset under another release.""" return make_dataset().metadata.model_copy( @@ -119,56 +125,32 @@ def pins(store): return _store_child_without_parent(store) -def test_build_release_builds_into_the_store_and_returns_the_plan(store, fake): - plan = build_release("a/b", store=store) - assert [(s.action, s.release) for s in plan.steps] == [("build", REF_AB)] - assert store.exists("a/b", "1.0.0") - assert fake.calls == [(REF_AB, False, {})] - - -def test_build_release_calls_the_claiming_backend_with_the_pins(store, fake_parent, fake_child, pins): - plan = build_release(CHILD_REF.dataset_id, store=store, build_parents=True) - # (release, force, pinned_by): the backend of another package gets the pins of the child. - assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})] - # The store holds the child, so the plan builds only its missing payload source. - assert fake_child.calls == [] - assert plan.steps == (Step(PARENT_REF, "build", pinned_by={CHILD_REF: pins}),) - assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version)) - - -def test_build_release_passes_the_pins_of_the_caller_to_the_backend(store, fake_parent, pins): - plan = build_release(PARENT_REF.dataset_id, store=store, pinned_by={CHILD_REF: pins}) +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 == (Step(PARENT_REF, "build", pinned_by={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_build_release_rejects_a_version_the_connector_does_not_declare(store, fake): +def test_plan_release_build_rejects_a_version_the_connector_does_not_declare(store, fake): with pytest.raises(TimeNetVersionUnavailableError) as caught: - build_release("a/b@2.0.0", store=store) + plan_release_build("a/b@2.0.0", store=store) assert (caught.value.release, caught.value.declared) == (DatasetRef.parse("a/b@2.0.0"), REF_AB.version) - assert str(caught.value) == ( - "connector for 'a/b' builds version 1.0.0, not the requested 2.0.0. " - "Ask for version 1.0.0, or leave out the version." - ) - assert fake.calls == [] - - -def test_build_release_accepts_the_version_that_the_connector_declares(store, fake): - build_release("a/b@1.0.0", store=store) - assert fake.calls == [(REF_AB, False, {})] -def test_build_release_copies_a_missing_parent_and_reports_each_step_once(tmp_path, store, fake_parent, fake_child): +def test_a_build_copies_a_missing_parent_and_reports_each_step_once(tmp_path, store, fake_parent, fake_child): origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) events = [] - plan = build_release( - CHILD_REF.dataset_id, store=store, fetch_from=origin, options=BuildOptions(progress=events.append) - ) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[origin]) + run_plan(plan, store, options=BuildOptions(progress=events.append)) - assert plan.steps == (Step(PARENT_REF, "copy"), Step(CHILD_REF, "build", depends_on=(PARENT_REF,))) + assert plan.steps == ( + CopyStep(PARENT_REF, origin), + _build(CHILD_REF, depends_on=(PARENT_REF,)), + ) assert events == [ StepStarted(PARENT_REF, "copy"), StepFinished(PARENT_REF, "copy"), @@ -179,31 +161,7 @@ def test_build_release_copies_a_missing_parent_and_reports_each_step_once(tmp_pa assert store.exists(PARENT_REF.dataset_id, str(PARENT_REF.version)) -@pytest.mark.usefixtures("fake_child") -@pytest.mark.parametrize("payload_source", ["missing", "damaged"]) -def test_build_release_reuses_a_held_release_and_copies_only_its_payload_source( - tmp_path, store, fake_parent, pins, payload_source -): - # The store holds the child, but not an intact copy of its payload source. The registry to copy - # from holds only the payload source. - origin = LocalRegistry(tmp_path / "origin") - origin.put_dataset(make_dataset()) - if payload_source == "damaged": - store.put_dataset(make_dataset()) - (release_dir(store.root, PARENT_REF) / "control.duckdb").write_bytes(b"cut short") - events = [] - - plan = build_release( - CHILD_REF.dataset_id, store=store, fetch_from=origin, options=BuildOptions(progress=events.append) - ) - - assert plan.steps == (Step(CHILD_REF, "copy"),) - assert events == [StepStarted(PARENT_REF, "copy"), StepFinished(PARENT_REF, "copy")] - assert fake_parent.calls == [] - assert complete_copy(release_dir(store.root, PARENT_REF), origin.get_manifest(PARENT_REF.dataset_id)) - - -def test_build_release_copies_a_release_whose_payload_source_only_the_store_holds( +def test_a_build_copies_a_release_whose_payload_source_only_the_store_holds( tmp_path, store, fake_parent, fake_child, backends ): # The registry to copy from holds the child, but not the parent whose files the child reads. @@ -213,15 +171,19 @@ def test_build_release_copies_a_release_whose_payload_source_only_the_store_hold fake_grandchild = _Backend(_grandchild_card()) backends.append(fake_grandchild) - plan = build_release(GRANDCHILD_REF.dataset_id, store=store, fetch_from=origin) + plan = plan_release_build(GRANDCHILD_REF.dataset_id, store=store, parents_from=[origin]) + run_plan(plan, store) - assert plan.steps == (Step(CHILD_REF, "copy"), Step(GRANDCHILD_REF, "build", depends_on=(CHILD_REF,))) + assert plan.steps == ( + CopyStep(CHILD_REF, origin), + _build(GRANDCHILD_REF, depends_on=(CHILD_REF,)), + ) assert fake_parent.calls == fake_child.calls == [] assert fake_grandchild.calls == [(GRANDCHILD_REF, False, {})] assert store.get_manifest(CHILD_REF.dataset_id) == origin.get_manifest(CHILD_REF.dataset_id) -def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( +def test_a_build_builds_a_payload_source_of_a_release_that_it_copies( tmp_path, store, fake_parent, fake_child, backends ): # The registry to copy from holds the child, but not the parent whose files the child reads. @@ -230,12 +192,13 @@ def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( fake_grandchild = _Backend(_grandchild_card()) backends.append(fake_grandchild) - plan = build_release(GRANDCHILD_REF.dataset_id, store=store, fetch_from=origin, build_parents=True) + plan = plan_release_build(GRANDCHILD_REF.dataset_id, store=store, parents_from=[origin, InstalledConnectors()]) + run_plan(plan, store) assert plan.steps == ( - Step(PARENT_REF, "build", pinned_by={CHILD_REF: pins}), - Step(CHILD_REF, "copy", depends_on=(PARENT_REF,)), - Step(GRANDCHILD_REF, "build", depends_on=(CHILD_REF,)), + _build(PARENT_REF, pinned_by={CHILD_REF: pins}), + CopyStep(CHILD_REF, origin, depends_on=(PARENT_REF,)), + _build(GRANDCHILD_REF, depends_on=(CHILD_REF,)), ) assert fake_parent.calls == [(PARENT_REF, False, {CHILD_REF: pins})] assert fake_child.calls == [] @@ -243,53 +206,48 @@ def test_build_release_builds_a_payload_source_of_a_release_that_it_copies( assert store.get_manifest(CHILD_REF.dataset_id) == origin.get_manifest(CHILD_REF.dataset_id) -def test_build_release_forces_only_the_requested_release(store, fake_parent, fake_child): +def test_a_forced_build_forces_only_the_requested_release(store, fake_parent, fake_child): store.put_dataset(_demo(_child_card())) - plan = build_release(CHILD_REF.dataset_id, store=store, build_parents=True, force=True) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()], force=True) + run_plan(plan, store) - assert [(s.action, s.release) for s in plan.steps] == [("build", PARENT_REF), ("build", CHILD_REF)] + assert plan.steps == (_build(PARENT_REF), _build(CHILD_REF, depends_on=(PARENT_REF,), force=True)) assert fake_parent.calls == [(PARENT_REF, False, {})] assert fake_child.calls == [(CHILD_REF, True, {})] -def test_build_release_skips_a_release_the_store_holds(store, fake): +def test_a_build_skips_a_release_the_store_holds(store, fake): + store.put_dataset(_demo(_card(REF_AB))) + assert plan_release_build("a/b", store=store).steps == () + + +def test_a_damaged_copy_of_the_requested_release_gets_a_forced_build(store, fake): store.put_dataset(_demo(_card(REF_AB))) - plan = build_release("a/b", store=store) - assert plan.steps == () - assert fake.calls == [] + (release_dir(store.root, REF_AB) / "control.duckdb").write_bytes(b"cut short") + + plan = plan_release_build("a/b", store=store) + + assert plan.steps == (_build(REF_AB, force=True),) + run_plan(plan, store) + assert fake.calls == [(REF_AB, True, {})] + assert store.damage(REF_AB) is None -def test_build_release_with_open_access_only_refuses_a_credentialed_card(store, backends): +def test_plan_release_build_with_open_access_only_refuses_a_credentialed_card(store, backends): fake_credentialed = _Backend(_card(REF_AB, **_GATED)) backends.append(fake_credentialed) with pytest.raises(TimeNetAccessError) as caught: - build_release("a/b", store=store, open_access_only=True) + plan_release_build("a/b", store=store, open_access_only=True) assert caught.value.dataset == REF_AB - assert fake_credentialed.calls == [] # Without the flag, the local store takes a release that is not open. - build_release("a/b", store=store) + run_plan(plan_release_build("a/b", store=store), store) assert fake_credentialed.calls == [(REF_AB, False, {})] -@pytest.mark.parametrize("held", [False, True]) -def test_build_release_with_open_access_only_refuses_a_credentialed_parent(store, backends, held): - gated = _Backend(make_dataset().metadata.model_copy(update=_GATED)) - child = _Backend(_child_card()) - backends.extend([gated, child]) - if held: - store.put_dataset(_demo(gated.cards[PARENT_REF.dataset_id])) - - with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD_REF.dataset_id, store=store, build_parents=True, open_access_only=True) - - assert caught.value.dataset == PARENT_REF - assert gated.calls == child.calls == [] - - -def test_build_release_with_open_access_only_refuses_a_held_release_with_a_credentialed_ancestor(store, backends): +def test_plan_release_build_with_open_access_only_refuses_a_held_release_with_a_credentialed_ancestor(store, backends): # The card of the child is open. The child keeps a Record of the open release a/b, which the store # lacks. It reads nothing from the credentialed parent, which is only in its lineage. child_card = _card(CHILD_REF, parents={ParentAlias("parent"): PARENT_REF, ParentAlias("open"): REF_AB}) @@ -312,12 +270,37 @@ def test_build_release_with_open_access_only_refuses_a_held_release_with_a_crede assert list(held.payload_sources) == [SourceKey(str(REF_AB))] with pytest.raises(TimeNetAccessError) as caught: - build_release(CHILD_REF.dataset_id, store=store, build_parents=True, open_access_only=True) + plan_release_build( + CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()], open_access_only=True + ) assert caught.value.dataset == CHILD_REF - assert other.calls == child.calls == [] -def test_build_release_without_a_backend_raises_not_found(store, backends): - with pytest.raises(TimeNetDatasetNotFoundError, match="no installed connector builds 'a/b'"): - build_release("a/b", store=store) +def test_a_plan_runs_only_through_run_plan(store, fake_parent, fake_child): + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()]) + + assert plan.steps == ( + _build(PARENT_REF), + _build(CHILD_REF, depends_on=(PARENT_REF,)), + ) + assert not store.root.exists() + assert fake_parent.calls == fake_child.calls == [] + run_plan(plan, store) + assert fake_parent.calls == [(PARENT_REF, False, {})] + assert fake_child.calls == [(CHILD_REF, False, {})] + + +def test_a_plan_reads_the_card_of_each_connector_once(store, backends): + reads = [] + + class _Counted(_Backend): + def card(self, dataset_id): + reads.append(dataset_id) + return super().card(dataset_id) + + backends.extend([_Counted(make_dataset().metadata), _Counted(_child_card())]) + plan = plan_release_build(CHILD_REF.dataset_id, store=store, parents_from=[InstalledConnectors()]) + run_plan(plan, store) + assert reads == [CHILD_REF.dataset_id, PARENT_REF.dataset_id] + assert store.exists(CHILD_REF.dataset_id, str(CHILD_REF.version)) diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py index a82858fe6..863fdb7d4 100644 --- a/packages/timenet/tests/registry/test_copy.py +++ b/packages/timenet/tests/registry/test_copy.py @@ -290,9 +290,13 @@ def test_copy_release_checks_the_access_of_a_payload_source_that_only_dest_holds def test_copy_release_refuses_a_payload_source_that_no_registry_holds(source, dest): + origin = _without_parent(source) with pytest.raises(TimeNetMissingParentError) as raised: - copy_release(child_ref, _without_parent(source), dest) + copy_release(child_ref, origin, dest) assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + # The error names both registries that the copy looked in. + assert raised.value.searched == f"{origin} or {dest}" + assert str(raised.value) == f"payload source {parent_ref} of {child_ref} is not in {origin} or {dest}." assert not dest.exists(*child) @@ -301,9 +305,11 @@ def test_copy_release_refuses_a_payload_source_that_only_dest_holds_damaged(sour copy_release(parent_ref, source, dest) parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version) (parent_dir / dest.get_manifest(*parent).files.control[0].path).unlink() + origin = _without_parent(source) with pytest.raises(TimeNetMissingParentError) as raised: - copy_release(child_ref, _without_parent(source), dest) + copy_release(child_ref, origin, dest) assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + assert raised.value.searched == f"{origin} or {dest}" assert not dest.exists(*child) diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 2307858d7..d75e84864 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -25,6 +25,7 @@ from timenet.manifest.files import FilePart from timenet.reader import Loading, LoadPlan, TimeFReader from timenet.registry import TIMENET_REGISTRY_URL, LocalRegistry, RemoteRegistry, open_local_version +from timenet.sources import InstalledConnectors from timenet.testing import assert_datasets_equal, install_backends, make_dataset from timenet.types import Access, DatasetMetadata, Domain, License, Version from timenet.types.access import AccessRequirement @@ -83,19 +84,19 @@ def build(self, release, store, *, force, pinned_by, options): def _record_build_requests(monkeypatch): - """Record the dataset, ``fetch_from`` and ``build_parents`` of each build that the client asks for. + """Record the dataset and the ``parents_from`` list of each build that the client asks for. Returns: - The list that receives one ``(dataset, fetch_from, build_parents)`` entry for each request. + The list that receives one ``(dataset, parents_from)`` entry for each request. """ requests = [] - build_release = client_module.build_release + plan_release_build = client_module.plan_release_build - def record(dataset, **options): - requests.append((dataset, options.get("fetch_from"), options.get("build_parents"))) - return build_release(dataset, **options) + def record(dataset_id, **options): + requests.append((dataset_id, options.get("parents_from"))) + return plan_release_build(dataset_id, **options) - monkeypatch.setattr(client_module, "build_release", record) + monkeypatch.setattr(client_module, "plan_release_build", record) return requests @@ -136,7 +137,7 @@ def test_a_missing_version_is_copied_from_fetch_from_before_it_is_built(tmp_path assert (tmp_path / "local/test/child/1.0.0/manifest.json").is_file() return # The build gets the same registry to copy its own missing parents from. - assert requests == [("test/child", origin, True)] + assert requests == [("test/child", [origin, InstalledConnectors()])] assert calls == [(DatasetRef.parse("test/child@1.0.0"), False, {})] @@ -229,7 +230,9 @@ def test_a_missing_payload_source_builds_with_its_own_connector_against_the_pins assert signal.to_arrow().equals(make_dataset().records[0].signals[0].to_arrow()) # The source builds with its own connector, which prepares its parents in the same order, against # the pins of the child that reads it. - assert requests == [(str(reference), fetch_from, True)] + assert requests == [ + (str(reference), [InstalledConnectors()] if fetch_from is None else [fetch_from, InstalledConnectors()]) + ] assert calls == [(reference, False, {DatasetRef.parse("test/child@1.0.0"): pins})] From 989895ec3c98979b2059485b7bc4e12c06dd62cf Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 17:11:00 +0100 Subject: [PATCH 2/6] refactor: mark the mapping onto a list of sources as glue The comments in the connectors API and CLI that map fetch_from and build_parents onto a list of sources now start with "Glue:". Part of TNET-157. --- packages/timenet-connectors/src/timenet_connectors/api.py | 3 ++- .../timenet-connectors/src/timenet_connectors/builder/cli.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/timenet-connectors/src/timenet_connectors/api.py b/packages/timenet-connectors/src/timenet_connectors/api.py index 346ce2c43..9a9832a15 100644 --- a/packages/timenet-connectors/src/timenet_connectors/api.py +++ b/packages/timenet-connectors/src/timenet_connectors/api.py @@ -66,7 +66,8 @@ def build( # ruff: ignore[too-many-arguments] plan = plan_release_build( f"{dataset_id}@{version}" if version else dataset_id, store=store, - # Maps fetch_from and build_parents onto a list of sources for the coordinator. + # Glue: maps fetch_from and build_parents onto a list of sources, until the producer API + # replaces this function. parents_from=[ *([source] if source is not None else []), *([InstalledConnectors()] if build_parents else []), diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index fd8fd0bb5..741764d26 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -246,7 +246,8 @@ def _build_to( # ruff: ignore[too-many-arguments] plan = plan_release_build( dataset_id, store=store, - # Maps --fetch-from and --build-parents onto a list of sources for the coordinator. + # Glue: maps --fetch-from and --build-parents onto a list of sources, until the flags of the + # defaults replace them. parents_from=[ *([fetch_from] if fetch_from is not None else []), *([InstalledConnectors()] if build_parents else []), From 98bf69ccc2db3c9c4a091f31994d4c72cceaf97b Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 11:36:15 +0100 Subject: [PATCH 3/6] refactor: mark the glue as a TODO The glue comments of this change start with TODO(adrian), so a search finds each one that is left. Part of TNET-157. --- packages/timenet-connectors/src/timenet_connectors/api.py | 4 ++-- .../timenet-connectors/src/timenet_connectors/builder/cli.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/timenet-connectors/src/timenet_connectors/api.py b/packages/timenet-connectors/src/timenet_connectors/api.py index 9a9832a15..0ce46c8dc 100644 --- a/packages/timenet-connectors/src/timenet_connectors/api.py +++ b/packages/timenet-connectors/src/timenet_connectors/api.py @@ -66,8 +66,8 @@ def build( # ruff: ignore[too-many-arguments] plan = plan_release_build( f"{dataset_id}@{version}" if version else dataset_id, store=store, - # Glue: maps fetch_from and build_parents onto a list of sources, until the producer API - # replaces this function. + # TODO(adrian): Remove temporary glue which maps fetch_from and build_parents onto a + # list of sources until the producer API replaces this function. parents_from=[ *([source] if source is not None else []), *([InstalledConnectors()] if build_parents else []), diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index 741764d26..e46444e07 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -246,8 +246,8 @@ def _build_to( # ruff: ignore[too-many-arguments] plan = plan_release_build( dataset_id, store=store, - # Glue: maps --fetch-from and --build-parents onto a list of sources, until the flags of the - # defaults replace them. + # TODO(adrian): Remove temporary glue which maps --fetch-from and --build-parents onto a + # list of sources until the flags of the defaults replace them. parents_from=[ *([fetch_from] if fetch_from is not None else []), *([InstalledConnectors()] if build_parents else []), From 368a7ee10238a5610821afe0d3cb915323ed80c8 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 19:00:42 +0100 Subject: [PATCH 4/6] docs: shorten the docstrings and docs of the build sources Say only what a caller of the build plan and of its list of sources needs to know. Part of TNET-157. --- packages/timenet/src/timenet/registry/_copy.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index ba07a0f23..09088027b 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -37,16 +37,14 @@ def copy_release( source: The registry to copy from. dest: The registry to copy to. with_payload_sources: Also copy the payload sources that the manifest of the release lists. - progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied or repaired release. + progress: Receives a ``StepStarted`` and a ``StepFinished`` event for each copied release. Raises: - TimeNetVersionConflictError: If ``dest`` holds the release or a payload source with other files or another card. - TimeNetAccessError: If a release lacks open access where the source or destination requires it. + TimeNetVersionConflictError: If ``dest`` holds a release of the copy with other files or another card. + TimeNetAccessError: If a release is not open, and ``dest`` or ``source`` requires open access. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. - TimeNetMissingParentError: If a payload source is neither in ``source`` nor complete in - ``dest``. The error names both registries. - TimeFFormatError: If a manifest names another release, or a payload source differs from the - files that the release pins. + TimeNetMissingParentError: If neither registry holds a complete payload source. + TimeFFormatError: If a manifest names another release or a pin fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload From 20513ef13795026631809af25d0f62fdea61e58e Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 00:21:20 +0100 Subject: [PATCH 5/6] docs: shorten the docstrings of the build sources Shorter docstrings for the code that this change adds or changes. Part of TNET-157. --- .../src/timenet_connectors/api.py | 32 +++++++++---------- .../src/timenet_connectors/builder/cli.py | 10 +++--- packages/timenet/src/timenet/client.py | 19 ++++++----- .../timenet/src/timenet/registry/_copy.py | 16 +++++----- 4 files changed, 38 insertions(+), 39 deletions(-) diff --git a/packages/timenet-connectors/src/timenet_connectors/api.py b/packages/timenet-connectors/src/timenet_connectors/api.py index 0ce46c8dc..4674cdeda 100644 --- a/packages/timenet-connectors/src/timenet_connectors/api.py +++ b/packages/timenet-connectors/src/timenet_connectors/api.py @@ -27,25 +27,25 @@ def build( # ruff: ignore[too-many-arguments] # ruff: ignore[docstring-extraneous-exception] (raised by the plan and its run, not directly here) """Build a dataset release in a local registry. - The build reuses an existing release unless you set ``force``. The connector runs in an environment - with its required packages, or in this interpreter if ``TIMENET_ISOLATION=off``. + The build reuses an existing release unless ``force`` is set. The connector runs in an environment + from its ``requirements.txt``, or in this interpreter if ``TIMENET_ISOLATION=off``. Args: - dataset_id: The connector's dataset ID (``org/name``). - version: The version to build, or ``None`` for the connector's declared version. - out: The output directory, or ``None`` to use the default local registry. - force: Rebuild the release even if it exists in the registry. - keep_cache: Keep downloaded source files after a successful build. - fetch_from: The registry, or its URI, to copy missing parent releases from. - build_parents: Build missing parent releases from their installed connectors. + dataset_id: The ``org/name`` ID declared in the connector's card. + version: The required card version, or ``None`` to accept any version. + out: The local registry directory, or ``None`` to use the default local registry. + force: Rebuild an existing release. + keep_cache: Keep the raw download cache after a successful build. + fetch_from: The registry or URI to copy missing parents from before trying builds. + build_parents: Build missing parents with installed connectors. Returns: - The release's version directory. + The version directory of the release. Raises: TimeNetDatasetNotFoundError: If no installed connector builds ``dataset_id``. - TimeNetVersionUnavailableError: If ``version`` differs from the connector's declared version. - TimeNetMissingParentError: If a required parent release is still missing. + TimeNetVersionUnavailableError: If the card's version differs from ``version``. + TimeNetMissingParentError: If a required release cannot be copied or built. """ from timenet.builders import BuildOptions, declared # ruff: ignore[import-outside-top-level] from timenet.coordinator.build import plan_release_build, run_plan # ruff: ignore[import-outside-top-level] @@ -82,14 +82,14 @@ def build( # ruff: ignore[too-many-arguments] def _missing_hint(*, fetch_from: "str | BaseRegistry | None", build_parents: bool) -> str | None: - """Name the keywords of :func:`build` that can get a missing release and that the call did not use. + """Suggest options to obtain a missing release. Args: - fetch_from: The value of ``fetch_from``, or ``None``. - build_parents: The value of ``build_parents``. + fetch_from: The selected registry or URI, or ``None``. + build_parents: Whether parent builds are enabled. Returns: - The hint as a full sentence, or ``None`` if the call used both keywords. + A sentence naming unused options, or ``None`` if both are enabled. """ if fetch_from is None and not build_parents: return "Copy it with fetch_from= or build it with build_parents=True." diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index e46444e07..a49ac0511 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -177,14 +177,14 @@ def build( def _missing_hint(*, fetch_from: str | None, build_parents: bool) -> str | None: - """Name the options of ``build`` that can get a missing release and that the command did not use. + """Suggest flags to obtain a missing release. Args: fetch_from: The value of ``--fetch-from``, or ``None``. build_parents: Whether ``--build-parents`` is set. Returns: - The hint as a full sentence, or ``None`` if the command used both options. + A sentence naming unused flags, or ``None`` if both are set. """ ways = [] if fetch_from is None: @@ -214,12 +214,12 @@ def _build_to( # ruff: ignore[too-many-arguments] dataset_id: The dataset ID. target: The local directory or remote registry to write to. fetch_from: The registry to copy missing parents from. ``None`` uses a remote output, if any. - build_parents: Build missing parents with their installed connectors. - force: Rebuild the release even if the local registry holds it. + build_parents: Build missing parents with installed connectors. + force: Rebuild an existing local release. options: Settings for each build. Returns: - The version directory for a local output, or the version string for a remote output. + The local version directory or the remote version string. Raises: BadParameter: If ``force`` is set and the remote output already holds the version. diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index d39839f25..cf2dca4d2 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -394,18 +394,17 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str | def _prepare( registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None ) -> DatasetVersion: - """Prepare a local dataset version and its payload sources. + """Prepare a local dataset version and the releases that supply its payload files. - This function reuses local releases, tries ``fetch_from``, then builds missing releases if - ``auto_build`` is true. Each rebuilt payload source must match the version and file pins the - dataset requires. + The function copies missing releases from ``fetch_from`` or builds them when ``auto_build`` is set. + A payload source build must reproduce the exact version and pinned files. Returns: The release with each payload source attached. Raises: - TimeNetMissingParentError: If copying and building cannot provide a missing payload source. - TimeNetInvalidCardError: If reading a connector's card fails. + TimeNetMissingParentError: If a missing payload source cannot be copied or built. + TimeNetInvalidCardError: If a connector's card cannot be read. """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and the build try: manifest = registry.get_manifest(dataset_id, version) @@ -447,17 +446,17 @@ def _obtain( # ruff: ignore[too-many-arguments] auto_build: bool, fetch_from: BaseRegistry | None, ) -> Manifest: - """Get a missing version's manifest from another registry or a local build. + """Get a missing version's manifest from ``fetch_from``, or build the version locally. - Try ``fetch_from`` first, then build if ``auto_build`` is true. + A local build needs ``auto_build`` and an installed connector for the dataset. Returns: - The manifest from ``fetch_from`` without copying its files, or the local manifest after a build. + The manifest from ``fetch_from`` without its files, or the local manifest after a build. Raises: TimeNetDatasetNotFoundError: If neither ``fetch_from`` nor a connector provides the version. TimeNetVersionUnavailableError: If the connector declares a different version. - TimeNetInvalidCardError: If reading the connector's card fails. + TimeNetInvalidCardError: If the connector's card cannot be read. """ # ruff: ignore[docstring-extraneous-exception] - raised by the build if fetch_from is not None: try: diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index 09088027b..db7ad0b6f 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -27,24 +27,24 @@ def copy_release( with_payload_sources: bool = True, progress: ProgressCallback | None = None, ) -> None: - """Copy a release and its payload sources to another registry. + """Copy a release and its required payload sources to another registry. - Copy payload sources first, reuse complete releases in ``dest``, and fetch damaged local copies again. - All checks run before the first copy, so a failed check leaves ``dest`` unchanged. + The copy reuses complete releases in ``dest`` and fetches damaged local copies again. + It checks access, file records and pins before copying payload sources and then the requested release. Args: reference: The exact release to copy. source: The registry to copy from. dest: The registry to copy to. with_payload_sources: Also copy the payload sources that the manifest of the release lists. - progress: Receives a ``StepStarted`` and a ``StepFinished`` event for each copied release. + progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied release. Raises: - TimeNetVersionConflictError: If ``dest`` holds a release of the copy with other files or another card. - TimeNetAccessError: If a release is not open, and ``dest`` or ``source`` requires open access. + TimeNetVersionConflictError: If ``dest`` holds other files or another card for a release being copied. + TimeNetAccessError: If a release fails an open access requirement of ``dest`` or ``source``. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. - TimeNetMissingParentError: If neither registry holds a complete payload source. - TimeFFormatError: If a manifest names another release or a pin fails. + TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it. + TimeFFormatError: If a manifest names a different release or a pin fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload From fef6ac79d49d78cb9bdc62cc5ce89bcd9fd922c4 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 10:21:41 +0100 Subject: [PATCH 6/6] test: trim the tests of the build sources Drop tests that cannot fail or repeat other tests, and sharpen the rest. Part of TNET-157. --- packages/timenet/tests/registry/test_copy.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py index 863fdb7d4..43514741f 100644 --- a/packages/timenet/tests/registry/test_copy.py +++ b/packages/timenet/tests/registry/test_copy.py @@ -305,11 +305,9 @@ def test_copy_release_refuses_a_payload_source_that_only_dest_holds_damaged(sour copy_release(parent_ref, source, dest) parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version) (parent_dir / dest.get_manifest(*parent).files.control[0].path).unlink() - origin = _without_parent(source) with pytest.raises(TimeNetMissingParentError) as raised: - copy_release(child_ref, origin, dest) + copy_release(child_ref, _without_parent(source), dest) assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) - assert raised.value.searched == f"{origin} or {dest}" assert not dest.exists(*child)