diff --git a/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py b/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py index 53e01c97e..2ccb73c89 100644 --- a/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py +++ b/packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py @@ -8,7 +8,7 @@ from timenet.dataset import TimeFDataset from timenet.errors import TimeFValidationError from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version from timenet.types import AnswerTask from timenet.types.splits import Split from timenet.types.text import read_value @@ -188,7 +188,7 @@ def test_splits_survive_storage_round_trip(tmp_path): original = TSQAConnector().convert(_refs(tmp_path, _balanced_rows())) registry = LocalRegistry(tmp_path / "registry") registry.put_dataset(original) - with TimeFReader(registry.open_version("chengsenwang/tsqa", "1.2.0")) as reader: + with TimeFReader(open_local_version(registry.root / "chengsenwang/tsqa/1.2.0")) as reader: restored = reader.read() assert {read_value(task.prompt): task.split for task in restored.tasks} == { read_value(task.prompt): task.split for task in original.tasks diff --git a/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py b/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py index 93f4c09ba..f97e78213 100644 --- a/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py +++ b/packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py @@ -13,7 +13,7 @@ from timenet.errors import TimeFFormatError, TimeFValidationError from timenet.json import JsonValue from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version from timenet.types import AnswerTask, ClassificationTask, Split from timenet_connectors.bases.huggingface import HubFileRef from timenet_connectors.datasets.tsaqa.tsaqa_benchmark import connector @@ -127,7 +127,7 @@ def test_storage_round_trip_keeps_the_whole_conversion(tmp_path): registry = LocalRegistry(tmp_path / "registry") registry.put_dataset(original) published = _published() - with TimeFReader(registry.open_version("tsaqa/tsaqa-benchmark", "1.0.0")) as reader: + with TimeFReader(open_local_version(registry.root / "tsaqa/tsaqa-benchmark/1.0.0")) as reader: restored = reader.read() assert {a.key: a.value for a in restored.annotations if a.key not in {"task", "question_type"}} == { "source_repository": "TSAQA/TSAQA-Benchmark", diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index e61241255..531d73db3 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -20,7 +20,7 @@ TimeNetMissingParentError, ) from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry, WritableRegistry +from timenet.registry import LocalRegistry, WritableRegistry, open_local_version from timenet.registry._paths import release_dir from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset @@ -176,7 +176,7 @@ def _child_of(registry: LocalRegistry, reference: DatasetRef, parent: DatasetRef update={"dataset_id": reference.dataset_id, "parents": {ParentAlias("parent"): parent}} ) child = TimeFDataset(metadata=metadata) - with TimeFReader(registry.open_version(parent.dataset_id, str(parent.version))) as reader: + with TimeFReader(open_local_version(release_dir(registry.root, parent))) as reader: child.import_from(reader, records=[record_id]) own = Signal.from_values( np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16) @@ -435,8 +435,8 @@ def get_manifest(self, dataset_id, version=None): def open_file(self, dataset_id, version, relpath): return self._local.open_file(dataset_id, version, relpath) - def open_version(self, dataset_id, version=None): - return self._local.open_version(dataset_id, version) + def open_release(self, reference, **options): + return self._local.open_release(reference, **options) def put_dataset(self, dataset, **options): return self._local.put_dataset(dataset, **options) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index cf2dca4d2..da7ece6d4 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -17,6 +17,7 @@ from timenet.builders import installed_builder from timenet.config import settings from timenet.coordinator.build import plan_release_build, run_plan +from timenet.coordinator.read import read_release from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, @@ -256,7 +257,7 @@ def open( "fetch_from copies releases into a local registry; a remote client reads from its own registry" ) self._reject_unhosted_access(dataset_id, version) - return TimeFReader(registry.open_version(dataset_id, version)) + return TimeFReader(read_release(dataset_id, version, sources=[registry], store=None)) source = open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from try: return TimeFReader(_prepare(registry, dataset_id, version, auto_build=auto_build, fetch_from=source)) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 9aa22fdd5..b62364dbc 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -4,7 +4,7 @@ from typing import assert_never from timenet.builders import BuildOptions, declared -from timenet.coordinator.plan import BuildStep, CopyStep, Plan, plan_steps +from timenet.coordinator.plan import BuildStep, CopyStep, Plan, ReadStep, plan_steps from timenet.manifest import PayloadSource from timenet.progress import StepFinished, StepStarted from timenet.registry import copy_release @@ -58,21 +58,26 @@ def plan_release_build( # ruff: ignore[too-many-arguments] def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None: - """Run a plan's copy and build steps in order. + """Run the copy and build steps of a plan in order. + + Skip read steps. Args: - plan: The steps to run in dependency order. + plan: The steps to run. store: The local registry to copy and build into. - options: Build settings, including the progress callback for copies and builds. + options: Build settings, including the progress callback. Raises: - TimeNetVersionConflictError: If ``store`` holds different files for a copied release. - TimeFFormatError: If a build fails to reproduce pinned files. + TimeNetVersionConflictError: If a copied release conflicts with the files in ``store``. + TimeFFormatError: If a build changes files pinned by other releases. 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 ReadStep(): + # A read opens the release where it is, so it has nothing to run. + pass case CopyStep(): copy_release( step.release, diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index d86ac72f2..e9f0d7bf7 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -1,7 +1,8 @@ """Plan release copies and builds with dependencies first, before any connector runs.""" -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field, replace +from functools import cached_property from graphlib import CycleError, TopologicalSorter from typing import ClassVar, Literal, TypeAlias @@ -22,7 +23,9 @@ validate_manifest_reference, ) from timenet.registry.local import LocalRegistry +from timenet.registry.version import DatasetVersion from timenet.sources import ( + Found, FoundInConnector, FoundInRegistry, InstalledConnectors, @@ -37,6 +40,19 @@ from timenet.types.version import Version +@dataclass(frozen=True) +class ReadStep: + """A planned read of one release in the registry that holds it.""" + + release: DatasetRef + registry: BaseRegistry + """The registry to read from.""" + depends_on: tuple[DatasetRef, ...] = () + """The releases with steps that must finish before this step runs.""" + action: ClassVar[Literal["read"]] = "read" + """The kind of step, for display.""" + + @dataclass(frozen=True) class CopyStep: """A planned copy of one release from a registry into the local registry.""" @@ -72,13 +88,13 @@ class BuildStep: """The kind of step, for display.""" -Step: TypeAlias = CopyStep | BuildStep -"""A planned copy or build of one release.""" +Step: TypeAlias = ReadStep | CopyStep | BuildStep +"""A planned read, copy, or build of one release.""" @dataclass(frozen=True) class Plan: - """The copy and build steps for a dataset, in execution order.""" + """The steps of a build or read in execution order.""" steps: tuple[Step, ...] @@ -94,70 +110,226 @@ class Plan: @dataclass(frozen=True) class _Release: - """A required release with its role, planned step and dependencies.""" + """A required release with its role, its need, its planned step and its dependencies.""" role: Role """The release's role as a dataset or a source of payload files.""" + needed_in_local_registry: bool + """True if the plan needs a complete copy in the output, False if a source that serves the release is enough.""" step: Step | None - """The copy or build step before dependencies are set, or ``None`` for an existing release.""" + """The step before dependencies are set, or ``None`` for a release that the output holds.""" dependencies: tuple[DatasetRef, ...] """The parent or payload source releases that must be ready first.""" +@dataclass +class Lookups: + """A cache of source answers and release manifests for one plan. + + The cache stays valid only while no step runs, so use one instance for one plan only. + """ + + answers: dict[tuple[int, DatasetRef], ReleaseLookup] = field(default_factory=dict) + """Source answers keyed by source identity and release. + + Keep each source alive while its answers are cached. + """ + manifests: dict[tuple[int, DatasetRef], Manifest | None] = field(default_factory=dict) + """Cached manifests keyed by registry identity and release, with ``None`` for missing releases.""" + + def resolve(self, sources: Sequence[ReleaseSource], dataset_id: str, version: str | None) -> ReleaseLookup: + """Resolve a release from ordered sources and cache answers for exact versions. + + Args: + sources: The sources to search, in order. + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + + Returns: + The selected ``Found`` answer, otherwise the first ``NotServed`` answer or ``NotFound()``. + """ + return resolve_release(sources, dataset_id, version, ask=self._ask) + + def ask(self, source: ReleaseSource, reference: DatasetRef) -> ReleaseLookup: + """Return a cached answer for an exact release, or ask the source and cache it. + + Args: + source: The source to query. + reference: The exact release. + + Returns: + The source's answer for the release. + """ + 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 manifest(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: + """Return a cached release manifest, or read and cache it. + + A manifest in a cached source answer takes precedence over a registry read. + + Returns: + The manifest, or ``None`` if the release is absent from the registry. + """ + 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 _ask(self, source: ReleaseSource, dataset_id: str, version: Version | None) -> ReleaseLookup: + """Return the answer of a source for a release, from the cache for an exact version.""" + if version is not None: + return self.ask(source, DatasetRef(dataset_id=dataset_id, version=version)) + answer = source.resolve(dataset_id, None) + if isinstance(answer, Found): + # A later request for the exact version gets the same answer. + self.answers.setdefault((id(source), DatasetRef(dataset_id=dataset_id, version=answer.version)), answer) + return answer + + @dataclass class _Graph: - """A graph of required releases and their copy or build steps.""" + """A graph of required releases and their read, copy or build steps.""" - output: LocalRegistry + output: LocalRegistry | None + """The local registry for copies and builds, or ``None`` for a read that copies and builds nothing.""" sources: Sequence[ReleaseSource] - """Sources for missing releases in preference order, excluding the output.""" + """The sources in preference order. A read reads the output like any other source.""" open_only: bool """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.""" 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.""" + lookups: Lookups = field(default_factory=Lookups) + """The cache of source answers and release manifests.""" 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: + @cached_property + def build_sources(self) -> list[ReleaseSource]: + """The sources without the output, which is never a source of its own copies and builds.""" + output = self.output + return [source for source in self.sources if output is None or not _is_same_registry(source, output)] + + def visit( + self, + reference: DatasetRef, + needed_by: DatasetRef | None, + pinned: PayloadSource | None = None, + *, + needed_in_local_registry: bool, + answer: ReleaseLookup | None = None, + ) -> None: """Add a release and its dependencies to the graph. - The graph revisits an existing entry only when its role changes from payload source to dataset. + A node is planned again when a later visit needs more of it: a dataset instead of a payload source, or a + copy in the output instead of a read. + + Args: + reference: The exact release to add. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + pinned: File requirements from ``needed_by``, or ``None`` for a parent dataset. + needed_in_local_registry: True if the plan needs a complete copy in the output. + answer: The answer of the sources for ``reference``, or ``None`` to ask them. + """ + role: Role = "dataset" if pinned is None else "payload-source" + if pinned is not None and needed_by is not None: + self.pins.setdefault(reference, {})[needed_by] = pinned + found = self.releases.get(reference) + if found is not None: + role = "dataset" if "dataset" in {found.role, role} else "payload-source" + needed_in_local_registry = found.needed_in_local_registry or needed_in_local_registry + if (role, needed_in_local_registry) == (found.role, found.needed_in_local_registry): + return + if needed_in_local_registry: + self.local(reference, needed_by, role) + else: + self.read(reference, needed_by, role, answer) + + def read( + self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role, answer: ReleaseLookup | None + ) -> None: + """Plan a release that a source can serve, and plan its payload sources if it is a dataset. + + Args: + reference: The exact release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. + answer: The answer of the sources for ``reference``, or ``None`` to ask them. + + Raises: + TimeNetFileError: If a source other than a registry provides the release and the output holds a + damaged copy. + TimeFValidationError: If only installed connectors provide the release and the plan has no output. + TimeNetDatasetNotFoundError: If the registry of the answer no longer holds the release. + TimeNetAccessError: If a source lists the release but no source serves it. + TimeNetMissingParentError: If no source provides the payload source. + """ # ruff: ignore[docstring-extraneous-exception] - the file, access and missing errors are raised as values + if answer is None: + answer = self.lookups.resolve(self.sources, reference.dataset_id, str(reference.version)) + if isinstance(answer, FoundInRegistry): + manifest = self.lookups.manifest(answer.registry, reference) + if manifest is None: + raise TimeNetDatasetNotFoundError(f"{answer.registry} no longer holds {reference}") + # The check of the local files waits for the end of the walk, because a later copy can repair them. + sources = list(manifest.payload_sources.values()) if role == "dataset" else [] + step = ReadStep(reference, answer.registry) + self.releases[reference] = _Release(role, False, step, tuple(source.release for source in sources)) + for source in sources: + self.visit(source.release, reference, source, needed_in_local_registry=False) + return + # Nothing but a registry replaces a damaged copy in the output. + damage = self.in_output(reference)[1] + if damage is not None: + raise damage + if isinstance(answer, FoundInConnector): + if self.output is None: + raise TimeFValidationError( + f"{reference} needs a build by {InstalledConnectors()}, but the read has no local registry to " + "build into" + ) + self.local(reference, needed_by, role) + elif isinstance(answer, NotServed): + raise answer.access_error(reference.dataset_id) + else: + raise self.missing(reference, needed_by, role, self.sources) + + def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) -> None: + """Plan a complete copy of a release in the output, from the output itself, a copy or a build. Args: - reference: The release to add. - needed_by: The release that needs ``reference``. - pinned: File pins from ``needed_by``, or ``None`` for a parent dataset. + reference: The exact release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. Raises: + TimeFValidationError: If the plan has no output. 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.role == "dataset" or role == "payload-source"): - return + """ # ruff: ignore[docstring-extraneous-exception] - the file, access and missing errors are raised as values + output = self.output + if output is None: + raise TimeFValidationError(f"{reference} must be in a local registry, but the read has none") 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)) + self.releases[reference] = _Release(role, True, None, tuple(source.release for source in sources)) for source in sources: - self.visit(source.release, reference, source) + self.visit(source.release, reference, source, needed_in_local_registry=True) return - answer = self.resolve(reference) - if isinstance(answer, FoundInRegistry) and not _is_same_registry(answer.registry, self.output): + answer = self.lookups.resolve(self.build_sources, reference.dataset_id, str(reference.version)) + if isinstance(answer, FoundInRegistry): self.copy(reference, role, answer.registry) elif damage is not None: # A copy from another registry repairs a damaged copy. Nothing else replaces it. @@ -168,9 +340,7 @@ def visit(self, reference: DatasetRef, needed_by: DatasetRef, pinned: PayloadSou raise answer.access_error(reference.dataset_id) else: # 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" - ) + raise self.missing(reference, needed_by, role, (output, *self.build_sources)) 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. @@ -179,15 +349,15 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non lacks gets its own step first. Args: - reference: The release to copy. + reference: The exact 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. + TimeNetDatasetNotFoundError: If the release is absent from the registry. 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) + manifest = self.lookups.manifest(registry, reference) if manifest is None: raise TimeNetDatasetNotFoundError(f"{registry} no longer holds {reference}") if self.open_only: @@ -207,9 +377,9 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non 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)) + self.releases[reference] = _Release(role, True, step, tuple(sources.missing)) for source, pinned in sources.missing.items(): - self.visit(source, reference, pinned) + self.visit(source, reference, pinned, needed_in_local_registry=True) def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> None: """Add a build step and plan the parents declared in its card. @@ -223,43 +393,31 @@ def build(self, reference: DatasetRef, role: Role, builder: InstalledBuilder) -> if self.open_only: require_open_access(card) parents = tuple(card.parents.values()) - self.releases[reference] = _Release(role, BuildStep(reference, builder), parents) + self.releases[reference] = _Release(role, True, 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) + self.visit(parent, reference, needed_in_local_registry=True) - def resolve(self, reference: DatasetRef) -> ReleaseLookup: - """Find an exact release across the sources in preference order. + @staticmethod + def missing( + reference: DatasetRef, needed_by: DatasetRef | None, role: Role, searched: Iterable[ReleaseSource] + ) -> TimeNetDatasetNotFoundError: + """Create the error for a release that no source provides. 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. + reference: The missing release. + needed_by: The release that needs ``reference``, or ``None`` for the requested release. + role: Whether the release is a dataset or a payload source. + searched: The searched sources, in order. Returns: - The cached or newly fetched answer. + A ``TimeNetDatasetNotFoundError`` for the requested release, else a ``TimeNetMissingParentError``. """ - 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)) + if needed_by is None: + names = ", ".join(str(source) for source in searched) + return TimeNetDatasetNotFoundError(f"{reference} is in none of the sources: {names}.") + described = describe_searched_sources(searched, reference) + return TimeNetMissingParentError(reference, needed_by, described, payload_source=role == "payload-source") 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. @@ -267,35 +425,22 @@ def listed(self, registry: BaseRegistry, reference: DatasetRef) -> Manifest | No Returns: The manifest, or ``None`` if the registry does not hold the release. """ - if isinstance(self.ask(registry, reference), NotFound): + if isinstance(self.lookups.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] + return self.lookups.manifest(registry, reference) def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFileError | None]: - """Check a release in the output for missing files or incorrect file sizes. + """Find a release in the output and check local files for damage. + + Missing files and incorrect file sizes count as damage in a local registry. Returns: - ``(manifest, None)`` if intact, ``(None, error)`` if damaged, else ``(None, None)``. + ``(manifest, None)`` if found, ``(None, error)`` if damaged, or ``(None, None)`` if absent or if the + plan has no output. """ - manifest = self.manifest(self.output, reference) + if self.output is None: + return None, None + manifest = self.lookups.manifest(self.output, reference) if manifest is None: return manifest, None if reference not in self.damage: @@ -303,47 +448,51 @@ def in_output(self, reference: DatasetRef) -> tuple[Manifest | None, TimeNetFile 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. + """Check the pins of each pinned release, then the files of each release that a read opens in place. 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 + TimeFFormatError: If a read, 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 release of the plan. + TimeNetFileError: If a local file of a release that a read opens is missing or has another size. + """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins and open_release 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. + # A read or a copy uses the release in 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) + manifest = None if registry is None else self.lookups.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) + for reference, release in self.releases.items(): + step = release.step + if isinstance(step, ReadStep) and isinstance(step.registry, LocalRegistry): + # Opening a local release checks each of its files. + step.registry.open_release(reference, manifest=self.lookups.manifest(step.registry, reference)) + + def open(self, reference: DatasetRef) -> DatasetVersion: + """Open a release of the plan without its payload sources, after the steps of the plan ran. + + Returns: + The release in the registry of its read step, else in the output. + + Raises: + TimeFValidationError: If the release has no read step and the plan has no output. + """ + step = self.releases[reference].step + if isinstance(step, ReadStep): + return step.registry.open_release(reference, manifest=self.lookups.manifest(step.registry, reference)) + if self.output is None: + # Without an output, the planner plans only read steps. So this is a bug of the planner. + raise TimeFValidationError(f"{reference} has no read step and the plan of the read has no output") + # The steps changed the output, so the release opens with a manifest that is read again. + return self.output.open_release(reference) def order(self) -> list[DatasetRef]: """Order the releases so that each release comes after the releases that it depends on. @@ -426,27 +575,26 @@ def plan_steps( # ruff: ignore[too-many-arguments] open_access_only: bool, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> tuple[DatasetRef, list[Step]]: - """Plan a dataset build and its dependencies. + """Plan a dataset build and all required releases. A damaged copy of the requested release in the output gets a forced build. Args: builder: The installed connector that builds the dataset. - output: The local registry that the build writes to. - 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. + output: The registry to write built releases to. + parents_from: The sources to search for missing releases, in order. + force_build: Rebuild the requested release even if the output holds it. + open_access_only: Require open access for every release. + pinned_by: The releases that pin files of the requested release, with their file requirements. Returns: - The requested release and ordered steps, with its own build last if needed. + The requested release and ordered steps, ending with its build 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) root = builder.release + # The graph leaves the output out of the sources of its copies and builds. graph = _Graph(output, parents_from, open_only=open_access_only) if pinned_by: graph.pins[root] = dict(pinned_by) @@ -460,16 +608,14 @@ def plan_steps( # ruff: ignore[too-many-arguments] # closes a cycle, and the walk does not look it up again. if held is None: 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())) + graph.releases[root] = _Release("dataset", True, step, tuple(card.parents.values())) for parent in card.parents.values(): - graph.visit(parent, root) + graph.visit(parent, root, needed_in_local_registry=True) else: sources = list(held.payload_sources.values()) - graph.releases[root] = _Release( - role="dataset", step=None, dependencies=tuple(source.release for source in sources) - ) + graph.releases[root] = _Release("dataset", True, None, tuple(source.release for source in sources)) for source in sources: - graph.visit(source.release, root, source) + graph.visit(source.release, root, source, needed_in_local_registry=True) graph.check() return root, graph.create_ordered_steps() @@ -487,3 +633,23 @@ def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool: if source is output: return True return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve() + + +def describe_searched_sources(sources: Iterable[ReleaseSource], reference: DatasetRef) -> str: + """Describe the sources searched for a missing release. + + Args: + sources: The searched sources, in order. + reference: The missing release. + + Returns: + Source names joined with " or ", including connector versions that differ from the requested version. + """ + names = [] + for source in 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) diff --git a/packages/timenet/src/timenet/coordinator/read.py b/packages/timenet/src/timenet/coordinator/read.py new file mode 100644 index 000000000..a1a771529 --- /dev/null +++ b/packages/timenet/src/timenet/coordinator/read.py @@ -0,0 +1,120 @@ +"""Plan and read releases from ordered sources. + +A read opens each release in the registry that provides it, so a release and its payload sources can come from +different registries. A release that only installed connectors provide is built into the local registry first. +""" + +from collections.abc import Sequence +from typing import cast + +from timenet.builders import declared, find_card +from timenet.coordinator.build import run_plan +from timenet.coordinator.plan import Plan, ReadStep, _Graph, describe_searched_sources +from timenet.errors import TimeNetDatasetNotFoundError +from timenet.registry.dependencies import attach_payload_sources +from timenet.registry.local import LocalRegistry +from timenet.registry.version import DatasetVersion +from timenet.sources import InstalledConnectors, NotFound, ReleaseSource +from timenet.types.references import DatasetRef, is_latest + + +def plan_read( + dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None +) -> Plan: + """Plan and validate the steps needed to read a release. + + A release that a registry provides gets a ``"read"`` step, and a release that only installed connectors + provide gets the steps of its build into ``store``. This checks local files and payload source pins before + any step runs. + + Args: + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + sources: Registries and ``InstalledConnectors()`` to search, in order. + store: The local registry for copies and builds, or ``None`` to copy and build nothing. + + Returns: + The ordered plan, ending with the requested release's step if it has one. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + TimeNetVersionUnavailableError: If the installed connector builds a different version. + TimeNetAccessError: If a source lists a needed release but no source serves it. + TimeNetMissingParentError: If a required payload source or build parent is unavailable. + TimeNetFileError: If a required local copy has missing files or incorrect file sizes. + TimeFFormatError: If a payload source conflicts with a pin or two pins conflict. + TimeFValidationError: If a build has no local registry to build into, or the releases form a cycle. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the planner and the checks + graph, _ = _walk(dataset_id, version, sources=sources, store=store) + return Plan(tuple(graph.create_ordered_steps())) + + +def read_release( + dataset_id: str, + version: str | None, + *, + sources: Sequence[ReleaseSource], + store: LocalRegistry | None, +) -> DatasetVersion: + """Open a release with its payload sources after completing required copies and builds. + + This can raise any error listed by :func:`plan_read`. + A local registry is required only for copies and builds. + + Args: + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None``, as in :func:`~timenet.sources.resolve_release`. + sources: Registries and ``InstalledConnectors()`` to search, in order. + store: The local registry for copies and builds, or ``None`` to copy and build nothing. + + Returns: + The release with attached payload sources that match its file pins. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + TimeNetVersionConflictError: If a copied release conflicts with the files in ``store``. + TimeFFormatError: If a build changes files pinned by other releases. + TimeNetBuildError: If a build fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the plan, the run and the open + graph, release = _walk(dataset_id, version, sources=sources, store=store) + plan = Plan(tuple(graph.create_ordered_steps())) + if any(not isinstance(step, ReadStep) for step in plan.steps): + # The plan copies and builds only into a local registry. + run_plan(plan, cast("LocalRegistry", store)) + return attach_payload_sources( + graph.open(release), graph.open, searched=lambda: describe_searched_sources(sources, release) + ) + + +def _walk( + dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None +) -> tuple[_Graph, DatasetRef]: + """Plan a read in one walk of the release graph, and check the plan. + + Returns: + The graph, and the requested release with its exact version. + + Raises: + TimeNetDatasetNotFoundError: If no source provides the release. + """ + graph = _Graph(store, sources, open_only=False) + answer = graph.lookups.resolve(sources, dataset_id, version) + if isinstance(answer, NotFound): + requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}") + if requested is not None: + # Nothing but a registry replaces a damaged copy in the local registry. + damage = graph.in_output(requested)[1] + if damage is not None: + raise damage + connectors = any(isinstance(source, InstalledConnectors) for source in sources) + if connectors and find_card(dataset_id) is not None: + # The installed connector builds another version. + declared(dataset_id, requested.version) + names = ", ".join(str(source) for source in sources) + wanted = dataset_id if requested is None else str(requested) + raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.") + release = DatasetRef(dataset_id=dataset_id, version=answer.version) + # The walk takes the answer, so that a read without an exact version does not ask again for the exact version. + graph.visit(release, None, needed_in_local_registry=False, answer=answer) + graph.check() + return graph, release diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index eb9791782..aa4c0364f 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -2,7 +2,7 @@ A registry serves compiled TimeF versions. It does not run connector code. Concrete backends must implement the four data-access methods: :meth:`list_datasets`, :meth:`get_manifest`, -:meth:`open_file`, and :meth:`open_version`. :meth:`BaseRegistry.search` is shared. It filters +:meth:`open_file`, and :meth:`open_release`. :meth:`BaseRegistry.search` is shared. It filters the output of :meth:`list_datasets` and uses :meth:`get_manifest` for the type filters. """ @@ -35,7 +35,11 @@ class BaseRegistry(ABC): - """A registry that provides TimeF dataset metadata, manifests, and files.""" + """A searchable source of TimeF datasets, manifests, and files. + + :meth:`open_release` and :meth:`download_release` handle one release without its payload + sources. :func:`~timenet.registry.copy_release` copies a release with its payload sources. + """ def __str__(self) -> str: return type(self).__name__ @@ -109,24 +113,23 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """ @abstractmethod - def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: - """Open a committed dataset version as a random-access handle. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open an exact release without attaching its payload sources. - The returned :class:`~timenet.registry.version.DatasetVersion` object bundles the parsed - manifest with a handle to the files of the version. The handle is rooted in a filesystem. - A reader built from this object does not re-open the registry. It also does not re-read - ``manifest.json``. The filesystem of the handle MUST serve range reads. The reader uses it - to read Parquet footers and value slices out of order. + Opening reads metadata only. Attach payload sources with + :func:`~timenet.registry.dependencies.attach_payload_sources`. Args: - dataset_id: The dataset id. - version: The version string, or ``None`` for the latest version. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle to the manifest and files of the committed version. + A handle to the release's manifest and files. Raises: - TimeNetDatasetNotFoundError: If the dataset id or version is unknown. + TimeNetDatasetNotFoundError: If the registry lacks the release. + TimeFFormatError: If the manifest declares a different release. + TimeNetFileError: If a local file is missing or has an incorrect size. """ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 29b533bd4..280f0c275 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -104,11 +104,11 @@ def open_source(reference: DatasetRef) -> DatasetVersion: raise TimeNetDatasetNotFoundError(f"no manifest at {source_manifest}") return DatasetVersion(read_manifest(source_manifest), version.filesystem, str(source_dir)) - return attach_payload_sources(version, open_source, searched=str(root)) + return attach_payload_sources(version, open_source, searched=lambda: str(root)) def attach_payload_sources( - version: DatasetVersion, open_source: Callable[[DatasetRef], DatasetVersion], *, searched: str + version: DatasetVersion, open_source: Callable[[DatasetRef], DatasetVersion], *, searched: Callable[[], str] ) -> DatasetVersion: """Open a release's payload sources and check each against the files the release pins. @@ -119,7 +119,7 @@ def attach_payload_sources( version: The opened release, without its payload sources. open_source: Open one exact release, without its payload sources. It raises :class:`~timenet.errors.TimeNetDatasetNotFoundError` when the release is absent. - searched: Where ``open_source`` looks, for the error that names a missing source. + searched: Describe where ``open_source`` looks, only for the error that names a missing source. Returns: The release with each payload source attached. @@ -136,7 +136,7 @@ def attach_payload_sources( try: source = open_source(pinned.release) except TimeNetDatasetNotFoundError as error: - raise TimeNetMissingParentError(pinned.release, release, searched, payload_source=True) from error + raise TimeNetMissingParentError(pinned.release, release, searched(), payload_source=True) from error validate_manifest_reference(source.manifest, pinned.release) check_pins(release, pinned, source.manifest) check_file_sizes(source, pinned.files, owner=f"payload source {pinned.release} of {release}") @@ -324,39 +324,34 @@ def open_complete(ref: DatasetRef, *, requested: bool) -> DatasetVersion: return DatasetVersion(release, pafs.LocalFileSystem(), str(directory)) version = open_complete(reference, requested=True) - return attach_payload_sources(version, lambda ref: open_complete(ref, requested=False), searched=str(root)) + return attach_payload_sources(version, lambda ref: open_complete(ref, requested=False), searched=lambda: str(root)) -def open_cached_or_remote( - root: Path, - reference: DatasetRef, - open_remote_version: Callable[[DatasetRef], DatasetVersion], - *, - searched: str, +def open_cached_or_remote_release( + root: Path, reference: DatasetRef, open_remote_version: Callable[[DatasetRef], DatasetVersion] ) -> DatasetVersion: - """Open a release and its payload sources from complete cached copies or remotely. + """Open a complete cached release or fall back to remote storage. + + Payload sources remain unattached. Args: - root: The cache directory, checked for complete releases before using remote files. + root: The cache directory to check before opening remote files. reference: The exact dataset ID and version to open. open_remote_version: Open one remote release without its payload sources. - searched: The remote registry, for the error that names a missing source. Returns: - A local or remote handle with each payload source attached and checked against its pins. - No catalogs or payload files are opened here. - """ + A handle to the complete cached copy, or a remote handle if the cache is incomplete. - def open_one(ref: DatasetRef) -> DatasetVersion: - directory = release_dir(root, ref) - cached = _load_complete_local_manifest(directory, ref) - if cached is not None: - return DatasetVersion(cached, pafs.LocalFileSystem(), str(directory)) - return open_remote_version(ref) - - version = open_one(reference) + Raises: + TimeFFormatError: If the manifest declares a different release. + """ # ruff: ignore[docstring-extraneous-exception] - raised by validate_manifest_reference + directory = release_dir(root, reference) + cached = _load_complete_local_manifest(directory, reference) + if cached is not None: + return DatasetVersion(cached, pafs.LocalFileSystem(), str(directory)) + version = open_remote_version(reference) validate_manifest_reference(version.manifest, reference) - return attach_payload_sources(version, open_one, searched=f"{searched} or its cache {root}") + return version def _ensure_dataset_version_available_locally( # ruff: ignore[too-many-arguments] diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index dca91c38b..9368a596e 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -9,12 +9,12 @@ from timenet.dataset import TimeFDataset from timenet.errors import TimeFValidationError -from timenet.manifest import PayloadSource +from timenet.manifest import Manifest, PayloadSource from timenet.registry._paths import release_dir -from timenet.registry.dependencies import open_cached_or_remote +from timenet.registry.dependencies import open_cached_or_remote_release from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry -from timenet.types.references import DatasetRef, is_latest +from timenet.types.references import DatasetRef from timenet.writer import TimeFWriter, WriteProgressEvent @@ -81,29 +81,28 @@ def put_dataset( finally: shutil.rmtree(staging, ignore_errors=True) - def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: - """Open a release and its payload sources, then read remote files on demand. - - Opening reads only manifests. Each payload source must be in this registry and match the - files the release pins. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open a cached or remote release without attaching its payload sources. Args: - dataset_id: The dataset ID. - version: The release version, or ``None`` for the latest. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle that prefers complete cached releases and otherwise uses remote range reads. + A handle to a complete copy in :attr:`local_root`, or to remote files read on demand. """ - if is_latest(version): - version = str(self.get_manifest(dataset_id, version).metadata.dataset_version) - return open_cached_or_remote( - self.local_root, DatasetRef.parse(f"{dataset_id}@{version}"), self._open_remote_version, searched=str(self) + return open_cached_or_remote_release( + self.local_root, reference, lambda ref: self._open_remote_version(ref, manifest=manifest) ) @abstractmethod - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: - """Bind one remote release to a range-capable filesystem, without its payload sources. + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open a remote release on a filesystem that serves range reads. + + Args: + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle whose payload sources :meth:`open_version` attaches. + A handle to the remote release without attached payload sources. """ diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index e0a077263..ab4e8fb43 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -13,7 +13,7 @@ from timenet.format.constants import MANIFEST_FILE from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry._paths import release_dir -from timenet.registry.dependencies import check_file_sizes, open_local_release +from timenet.registry.dependencies import check_file_sizes, validate_manifest_reference from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup @@ -268,25 +268,29 @@ def damage(self, release: DatasetRef, manifest: Manifest | None = None) -> TimeN return error return None - def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: - """Open a committed version as a local, random-access handle. + def open_release(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open a local release without attaching its payload sources. - Reuses the manifest :meth:`get_manifest` already parsed and validated, so a missing version or a - misplaced artifact surfaces there. Roots the handle at the version's directory over a - :class:`pyarrow.fs.LocalFileSystem`, which is zero network and already seekable. Opening reads - only metadata, as :func:`~timenet.registry.open_local_version` does, and finds payload sources - in this registry. + This checks that every file exists with its recorded size. Args: - dataset_id: The dataset id. - version: The version string, or ``None`` for the latest. + reference: The dataset ID and exact version to open. + manifest: A manifest already read from this registry, or ``None`` to fetch it. Returns: - A handle to the committed version's manifest and files. - """ - manifest = self.get_manifest(dataset_id, version) - reference = DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version) - return open_local_release(release_dir(self._root, reference), manifest, sources_root=self._root) + A handle to the release's directory on the local filesystem. + + Raises: + TimeNetDatasetNotFoundError: If the registry lacks a committed manifest for the release. + TimeFFormatError: If the manifest declares a different release. + TimeNetFileError: If a file is missing or has an incorrect size. + """ # ruff: ignore[docstring-extraneous-exception] - raised by get_manifest and the checks + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + validate_manifest_reference(manifest, reference) + version = DatasetVersion(manifest, pafs.LocalFileSystem(), str(release_dir(self._root, reference))) + check_file_sizes(version, manifest.files.all_files(), owner=str(reference)) + return version def _dataset_dir(self, dataset_id: str) -> Path: """Return the directory holding a dataset's versions. diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index d33cece8f..ad093591c 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -32,7 +32,7 @@ class RemoteRegistry(LazyReadRegistry): - """Read and publish datasets through the hosted TimeNet HTTP service.""" + """A dataset registry backed by the hosted TimeNet HTTP service.""" def __init__( self, @@ -118,13 +118,18 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: buffer.seek(0) return buffer - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: - """Bind one remote manifest to a range-capable filesystem. + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open a remote release with on-demand file reads. + + Args: + reference: The dataset ID and exact version to open. + manifest: A manifest already fetched from this registry, or ``None`` to fetch it. Returns: - A handle without its payload sources, which ``open_version`` attaches. + A handle to the remote release without attached payload sources. """ - manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) root = f"{reference.dataset_id}/{reference.version}" files = {f"{root}/{part.path}": part.size for part in manifest.files.all_files()} filesystem = RegistryFileSystem(self._base_url, files, token=self._http.token, transport=self._http.transport) diff --git a/packages/timenet/src/timenet/registry/s3.py b/packages/timenet/src/timenet/registry/s3.py index 4ec31d639..4bdf97d5f 100644 --- a/packages/timenet/src/timenet/registry/s3.py +++ b/packages/timenet/src/timenet/registry/s3.py @@ -4,9 +4,9 @@ matches the hosted registry's bucket layout. The manifest's relative file paths do not change. Only the key prefix differs from a :class:`~timenet.registry.LocalRegistry`. Credentials, region, and an optional endpoint override come from boto3's default session (``AWS_*`` vars, ``AWS_PROFILE``, -``AWS_ENDPOINT_URL``). Opening a version reads its payload sources' manifests upfront and reads remote -payloads on demand. Complete cached releases are reused. The backend has no catalog, so listing and -search are unsupported. Use it to publish and to fetch by an explicit ``org/name@version``. +``AWS_ENDPOINT_URL``). Opening a version reads remote payloads on demand. Complete cached releases +are reused. The backend has no catalog, so listing and search are unsupported. Use it to publish and +to fetch by an explicit ``org/name@version``. boto3 is optional. Install the ``timenet[s3]`` extra to use this backend. """ @@ -38,7 +38,7 @@ class S3Registry(LazyReadRegistry): - """Serves and publishes datasets under an ``s3://bucket/prefix`` root.""" + """A registry that reads and writes datasets under an ``s3://bucket/prefix`` root.""" _staging_prefix = "timenet-s3-" @@ -140,13 +140,18 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: raise TimeNetDatasetNotFoundError(f"no object {relpath!r} for {dataset_id!r} version {version!r}") from exc return response["Body"] - def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: - """Bind one remote manifest to Arrow's native S3 filesystem. + def _open_remote_version(self, reference: DatasetRef, *, manifest: Manifest | None = None) -> DatasetVersion: + """Open a release through Arrow's S3 filesystem. + + Args: + reference: The dataset ID and exact version to open. + manifest: A manifest already fetched from this registry, or ``None`` to fetch it. Returns: - A handle without its payload sources, which ``open_version`` attaches. + A handle to the S3 release without attached payload sources. """ - manifest = self.get_manifest(reference.dataset_id, str(reference.version)) + if manifest is None: + manifest = self.get_manifest(reference.dataset_id, str(reference.version)) root = f"{self._bucket}/{self._key(reference.dataset_id, str(reference.version))}" return DatasetVersion(manifest, self._s3_filesystem(), root) diff --git a/packages/timenet/tests/_demo_child.py b/packages/timenet/tests/_demo_child.py new file mode 100644 index 000000000..4316831c2 --- /dev/null +++ b/packages/timenet/tests/_demo_child.py @@ -0,0 +1,32 @@ +"""A local registry with the demo release and a child that reads one of its Records.""" + +from pathlib import Path + +from timenet.dataset import TimeFDataset +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, open_local_version +from timenet.registry._paths import release_dir +from timenet.testing import make_dataset +from timenet.types.reference_keys import ParentAlias +from timenet.types.references import DatasetRef + + +def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: + """Store the demo dataset and ``test/child``, which reads one of its Records, in a local registry. + + Returns: + The registry, and the demo release, which is the payload source of the child. + """ + parent = make_dataset() + registry = LocalRegistry(root) + registry.put_dataset(parent) + reference = DatasetRef(dataset_id=parent.metadata.dataset_id, version=parent.metadata.dataset_version) + child = TimeFDataset( + metadata=parent.metadata.model_copy( + update={"dataset_id": "test/child", "parents": {ParentAlias("parent"): reference}} + ) + ) + with TimeFReader(open_local_version(release_dir(root, reference))) as reader: + child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) + registry.put_dataset(child) + return registry, reference diff --git a/packages/timenet/tests/registry/_fake_registry.py b/packages/timenet/tests/_fake_registry.py similarity index 100% rename from packages/timenet/tests/registry/_fake_registry.py rename to packages/timenet/tests/_fake_registry.py diff --git a/packages/timenet/tests/conftest.py b/packages/timenet/tests/conftest.py index 3e7a2ecf6..cc6be6f89 100644 --- a/packages/timenet/tests/conftest.py +++ b/packages/timenet/tests/conftest.py @@ -5,10 +5,16 @@ """ import os +from pathlib import Path +import sys from hypothesis import HealthCheck, settings +# Make the test helpers of this directory (e.g. ``_demo_child``) importable by name under pytest's +# importlib import mode, which does not add test dirs to ``sys.path``. +sys.path.insert(0, str(Path(__file__).parent)) + settings.register_profile("dev", max_examples=10) settings.register_profile("ci", max_examples=100, print_blob=True) settings.register_profile( diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index 103af8160..16e22d550 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -10,7 +10,7 @@ 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 import LocalRegistry, open_local_version from timenet.registry._paths import release_dir from timenet.sources import InstalledConnectors from timenet.testing import install_backends, make_dataset @@ -112,7 +112,7 @@ def _store_child_without_parent(registry): parent = make_dataset() registry.put_dataset(parent) child = TimeFDataset(metadata=_child_card()) - with TimeFReader(registry.open_version(PARENT_REF.dataset_id, str(PARENT_REF.version))) as reader: + with TimeFReader(open_local_version(release_dir(registry.root, PARENT_REF))) as reader: child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) registry.put_dataset(child) shutil.rmtree(registry.root / PARENT_REF.dataset_id) @@ -257,8 +257,8 @@ def test_plan_release_build_with_open_access_only_refuses_a_held_release_with_a_ store.put_dataset(_demo(_card(REF_AB))) dataset = TimeFDataset(metadata=child_card) with ( - TimeFReader(store.open_version(PARENT_REF.dataset_id, str(PARENT_REF.version))) as gated_reader, - TimeFReader(store.open_version(REF_AB.dataset_id, str(REF_AB.version))) as open_reader, + TimeFReader(open_local_version(release_dir(store.root, PARENT_REF))) as gated_reader, + TimeFReader(open_local_version(release_dir(store.root, REF_AB))) as open_reader, ): dataset.import_from(gated_reader, alias=ParentAlias("parent")) dataset.import_from(open_reader, alias=ParentAlias("open"), records=[make_dataset().records[0].id]) diff --git a/packages/timenet/tests/coordinator/test_read.py b/packages/timenet/tests/coordinator/test_read.py new file mode 100644 index 000000000..15324741e --- /dev/null +++ b/packages/timenet/tests/coordinator/test_read.py @@ -0,0 +1,578 @@ +from pathlib import Path +import re +import shutil + +from _demo_child import _child_of_the_demo +from _fake_registry import build_fake, build_service_fake +import httpx +import numpy as np +import pytest + +from timenet import builders +from timenet.coordinator.plan import BuildStep, CopyStep, ReadStep +from timenet.coordinator.read import plan_read, read_release +from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset +from timenet.errors import ( + TimeFFormatError, + TimeFValidationError, + TimeNetAccessError, + TimeNetDatasetNotFoundError, + TimeNetFileError, + TimeNetMissingParentError, + TimeNetVersionUnavailableError, +) +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, RemoteRegistry, open_local_version +from timenet.registry._paths import release_dir +from timenet.sources import InstalledConnectors, ReleaseSource +from timenet.testing import install_backends, make_dataset +from timenet.types import Access, TimeSeriesSpec, ureg +from timenet.types.access import AccessRequirement +from timenet.types.reference_keys import ParentAlias, SourceKey +from timenet.types.references import DatasetRef + + +DEMO = DatasetRef.parse("timenet/hello-world@1.0.0") +CHILD = DatasetRef.parse("test/child@1.0.0") +PARENT = DatasetRef.parse("test/parent@1.0.0") +OTHER = DatasetRef.parse("test/other@1.0.0") +_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( + update={"dataset_id": reference.dataset_id, "dataset_version": reference.version, **update} + ) + + +def _demo(card): + """Return the demo dataset under ``card``.""" + dataset = make_dataset() + dataset._metadata = card + return dataset + + +class _Backend: + """A builder backend of another package. + + It builds CHILD as ``_child_of_the_demo()`` does, and the demo dataset under each other card. A + build hands its pins to the writer, which publishes the release only if it reproduces them. + """ + + def __init__(self, cards, calls): + self.cards = {card.dataset_id: card for card in cards} + self.calls = calls + + def dataset_ids(self): + return list(self.cards) + + def card(self, dataset_id): + return self.cards.get(dataset_id) + + def build(self, release, store, *, force, pinned_by, options): + self.calls.append((release, force, dict(pinned_by))) + if release == CHILD: + _child_of_the_demo(store.root) + else: + store.put_dataset(_demo(self.cards[release.dataset_id]), force=force, pinned_by=pinned_by or None) + + +def _install_demo_backend(monkeypatch, *cards) -> list: + """Make one backend the only installed one. It builds the demo release, and a release for each card. + + Returns: + The list that receives ``(release, force, pinned_by)`` for each build. + """ + calls = [] + backend = _Backend((make_dataset().metadata, *cards), calls) + install_backends(monkeypatch, backend) + return calls + + +def _child_reading(store, *cards, child_release=CHILD): + """Store a child, which reads one Record of the demo dataset under each card, without those releases. + + Args: + store: The registry to store the child in. + *cards: The cards of the releases that the child reads. + child_release: The release of the child. + + Returns: + The pins of the child on each release that it reads, in the order of its manifest. + """ + parents = { + ParentAlias(f"parent{index}"): DatasetRef(dataset_id=card.dataset_id, version=card.dataset_version) + for index, card in enumerate(cards) + } + child = TimeFDataset(metadata=_card(child_release, parents=parents)) + for (alias, parent), card in zip(parents.items(), cards, strict=True): + store.put_dataset(_demo(card)) + with TimeFReader(open_local_version(release_dir(store.root, parent))) as reader: + child.import_from(reader, alias=alias, records=[make_dataset().records[0].id]) + store.put_dataset(child) + for parent in parents.values(): + shutil.rmtree(release_dir(store.root, parent)) + manifest = store.get_manifest(child_release.dataset_id) + return {pinned.release: pinned for pinned in manifest.payload_sources.values()} + + +def _gated_remote(tmp_path) -> RemoteRegistry: + """Return a remote registry that lists DEMO as credentialed, so it does not serve its files.""" + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(_demo(make_dataset().metadata.model_copy(update=_GATED))) + transport, _ = build_fake(release_dir(origin.root, DEMO)) + return RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") + + +def _offline(request): + raise httpx.ConnectError("the network is down", request=request) + + +def _count_reads(monkeypatch, registry) -> dict[str, int]: + """Count each manifest that is read from a registry, by the release that it was asked for.""" + 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_read_reads_in_place_from_the_first_source_that_holds_the_release(tmp_path): + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + plan = plan_read(DEMO.dataset_id, None, sources=[store, origin], store=store) + assert plan.steps == (ReadStep(DEMO, origin),) + version = read_release(DEMO.dataset_id, None, sources=[store, origin], store=store) + assert Path(version.root).is_relative_to(origin.root) + assert not store.root.exists() + + +def test_a_damaged_copy_fails_the_read_before_any_step_runs(tmp_path, monkeypatch): + store, _ = _child_of_the_demo(tmp_path / "store") + # No registry holds DEMO, so the plan builds it before the read opens CHILD. + shutil.rmtree(release_dir(store.root, DEMO)) + # Origin and the connector also provide CHILD, but the read stops at the damaged copy, which comes first. + origin = LocalRegistry(tmp_path / "origin") + shutil.copytree(store.root, origin.root) + (release_dir(store.root, CHILD) / "control.duckdb").write_bytes(b"cut short") + calls = _install_demo_backend(monkeypatch, _card(CHILD, parents={ParentAlias("parent"): DEMO})) + with pytest.raises(TimeNetFileError, match=r"^test/child@1\.0\.0: file 'control\.duckdb' has 9 bytes"): + read_release(CHILD.dataset_id, None, sources=[store, origin, InstalledConnectors()], store=store) + assert calls == [] + + +@pytest.mark.parametrize("answer", ["build", "not-found", "not-served"]) +def test_a_damaged_copy_in_the_store_fails_a_read_that_no_registry_serves(tmp_path, monkeypatch, answer): + # The store is not in the list, so no registry answers for its damaged copy. + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + (release_dir(store.root, DEMO) / "control.duckdb").write_bytes(b"cut short") + _install_demo_backend(monkeypatch) + sources: list[ReleaseSource] + if answer == "build": + sources = [InstalledConnectors()] + elif answer == "not-found": + sources = [LocalRegistry(tmp_path / "empty")] + else: + # The damage error comes before the access error of the registry that lists DEMO. + sources = [_gated_remote(tmp_path)] + with pytest.raises(TimeNetFileError, match=r"^timenet/hello-world@1\.0\.0: file 'control\.duckdb' has 9 bytes"): + plan_read(DEMO.dataset_id, "1.0.0", sources=sources, store=store) + + +def test_a_payload_source_resolves_on_its_own(tmp_path): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + plan = plan_read(CHILD.dataset_id, None, sources=[store, origin], store=store) + assert plan.steps == (ReadStep(DEMO, origin), ReadStep(CHILD, store, depends_on=(DEMO,))) + version = read_release(CHILD.dataset_id, None, sources=[store, origin], store=store) + assert Path(version.root).is_relative_to(store.root) + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(origin.root) + assert not store.exists(DEMO.dataset_id, str(DEMO.version)) + + +def test_a_payload_source_with_other_files_than_the_pins_fails_the_plan(tmp_path): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + origin = LocalRegistry(tmp_path / "origin") + # The same release with its values in another format: CHILD pins files that it does not have. + origin.put_dataset(make_dataset(), values_backend="zarr") + with pytest.raises(TimeFFormatError, match=r"^payload source timenet/hello-world@1\.0\.0 differs"): + plan_read(CHILD.dataset_id, None, sources=[store, origin], store=store) + + +def test_a_missing_payload_source_builds_against_the_pins_of_the_release(tmp_path, monkeypatch): + store, _ = _child_of_the_demo(tmp_path / "store") + shutil.rmtree(release_dir(store.root, DEMO)) + pins = store.get_manifest(CHILD.dataset_id).payload_sources[SourceKey(str(DEMO))] + calls = _install_demo_backend(monkeypatch) + sources = [store, InstalledConnectors()] + assert plan_read(CHILD.dataset_id, None, sources=sources, store=store).steps == ( + _build(DEMO, pinned_by={CHILD: pins}), + ReadStep(CHILD, store, depends_on=(DEMO,)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(DEMO, False, {CHILD: pins})] + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +def test_a_read_builds_a_release_that_only_a_connector_provides(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + calls = _install_demo_backend(monkeypatch, _card(CHILD, parents={ParentAlias("parent"): DEMO})) + sources = [store, InstalledConnectors()] + # The build of CHILD reads DEMO, which the store holds, so DEMO gets no step. + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + assert plan.steps == (_build(CHILD),) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(CHILD, False, {})] + assert Path(version.root).is_relative_to(store.root) + # A payload source without a step opens from the store. + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +@pytest.mark.parametrize(("version", "release"), [(None, "x/y"), ("1.0.0", "x/y@1.0.0")]) +def test_a_release_that_no_source_provides_names_every_source(tmp_path, monkeypatch, version, release): + store = LocalRegistry(tmp_path / "store") + _install_demo_backend(monkeypatch) + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + plan_read("x/y", version, sources=[store, InstalledConnectors()], store=store) + assert type(caught.value) is TimeNetDatasetNotFoundError + assert str(caught.value) == f"{release} is in none of the sources: {store}, installed connectors." + + +def test_a_missing_payload_source_names_the_version_that_the_connector_builds(tmp_path, monkeypatch): + origin, _ = _child_of_the_demo(tmp_path / "origin") + shutil.rmtree(release_dir(origin.root, DEMO)) + _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + with pytest.raises(TimeNetMissingParentError) as caught: + plan_read(CHILD.dataset_id, None, sources=[origin, InstalledConnectors()], store=None) + assert str(caught.value) == ( + f"payload source {DEMO} of {CHILD} is not in {origin} or installed connectors ({DEMO.dataset_id} builds 2.0.0)." + ) + + +def test_pins_that_the_store_copy_of_a_payload_source_breaks_fail_the_read_before_any_step(tmp_path, monkeypatch): + # Origin holds CHILD, which reads OTHER and DEMO. Only the connectors provide them. The store + # holds DEMO with its values in another format, so the plan builds nothing that checks the pins + # of CHILD on DEMO. The plan builds OTHER first. + origin, store = LocalRegistry(tmp_path / "origin"), LocalRegistry(tmp_path / "store") + _child_reading(origin, _card(OTHER), make_dataset().metadata) + store.put_dataset(make_dataset(), values_backend="zarr") + calls = _install_demo_backend(monkeypatch, _card(OTHER)) + with pytest.raises(TimeFFormatError, match=r"^payload source timenet/hello-world@1\.0\.0 differs"): + read_release(CHILD.dataset_id, None, sources=[origin, InstalledConnectors()], store=store) + assert calls == [] + + +def test_a_release_that_a_source_lists_without_serving_raises_the_access_error(tmp_path): + with pytest.raises(TimeNetAccessError) as caught: + read_release(DEMO.dataset_id, None, sources=[_gated_remote(tmp_path)], store=None) + assert caught.value.requirements == (AccessRequirement(DEMO, Access.CREDENTIALED, _GATED["access_url"]),) + + +def test_an_exact_version_that_the_connector_does_not_declare_raises_version_unavailable(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + _install_demo_backend(monkeypatch) + with pytest.raises(TimeNetVersionUnavailableError) as caught: + read_release(DEMO.dataset_id, "2.0.0", sources=[store, InstalledConnectors()], store=store) + assert (caught.value.release, caught.value.declared) == (DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"), DEMO.version) + + +def test_latest_without_the_connectors_in_the_list_reads_the_older_copy_and_builds_nothing(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + # The installed connector builds a newer version, but the list does not hold the connectors. + calls = _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + assert plan_read(DEMO.dataset_id, "latest", sources=[store], store=store).steps == (ReadStep(DEMO, store),) + version = read_release(DEMO.dataset_id, "latest", sources=[store], store=store) + assert version.manifest.metadata.dataset_version == DEMO.version + assert calls == [] + + +def test_a_read_without_a_version_stops_at_the_first_source_that_serves_it(tmp_path, monkeypatch): + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + # A later registry cannot be reached, and the installed connector builds a newer version. + down = RemoteRegistry("http://down.local", transport=httpx.MockTransport(_offline), cache_dir=tmp_path / "c") + calls = _install_demo_backend(monkeypatch, _card(DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0"))) + + version = read_release(DEMO.dataset_id, None, sources=[store, down, InstalledConnectors()], store=store) + + assert version.manifest.metadata.dataset_version == DEMO.version + assert calls == [] + + +def test_a_read_that_builds_reads_the_card_of_the_connector_once(tmp_path, monkeypatch): + reads = [] + + class _Counted(_Backend): + def card(self, dataset_id): + reads.append(dataset_id) + return super().card(dataset_id) + + install_backends(monkeypatch, _Counted((make_dataset().metadata,), [])) + plan_read(DEMO.dataset_id, None, sources=[InstalledConnectors()], store=LocalRegistry(tmp_path / "store")) + assert reads == [DEMO.dataset_id] + + +def test_a_read_that_finds_every_release_asks_the_connectors_nothing(tmp_path, monkeypatch): + asked = [] + + class _Counted(_Backend): + def dataset_ids(self): + asked.append("dataset_ids") + return super().dataset_ids() + + def card(self, dataset_id): + asked.append(dataset_id) + return super().card(dataset_id) + + install_backends(monkeypatch, _Counted((make_dataset().metadata,), [])) + store, _ = _child_of_the_demo(tmp_path / "store") + read_release(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) + # Only the error for a missing payload source names the sources, with the versions that the connectors build. + assert asked == [] + + +def test_a_read_that_needs_a_build_without_a_store_raises(monkeypatch): + _install_demo_backend(monkeypatch) + with pytest.raises(TimeFValidationError, match=r"timenet/hello-world@1\.0\.0 needs a build"): + read_release(DEMO.dataset_id, None, sources=[InstalledConnectors()], store=None) + + +@pytest.mark.parametrize("parent", [PARENT, DatasetRef.parse("x/parent@1.0.0")], ids=["build-first", "read-first"]) +def test_a_release_that_a_build_copies_is_read_from_the_store_and_appears_once(tmp_path, monkeypatch, parent): + # CHILD reads the parent and DEMO. Only the connectors build the parent, and its card declares + # DEMO, which origin holds. So the build of the parent copies DEMO into the store, and the read + # takes it there. The read plans the payload sources in the order of their names. + parent_card = _card(parent, parents={ParentAlias("demo"): DEMO}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, parent_card, make_dataset().metadata) + origin.put_dataset(make_dataset()) + calls = _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + reads = _count_reads(monkeypatch, origin) + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + # The read and the build of the parent share what the plan read. + assert reads == {str(DEMO): 1} + assert plan.steps == ( + CopyStep(DEMO, origin), + _build(parent, depends_on=(DEMO,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(parent, False, {CHILD: pins[parent]})] + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +def test_a_step_that_two_builds_need_appears_once(tmp_path, monkeypatch): + # CHILD reads PARENT and OTHER. Only the connectors build them, and both of their cards declare + # DEMO, which origin holds. So both builds copy DEMO first. + cards = [_card(reference, parents={ParentAlias("demo"): DEMO}) for reference in (PARENT, OTHER)] + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, *cards) + origin.put_dataset(make_dataset()) + _install_demo_backend(monkeypatch, *cards) + + plan = plan_read(CHILD.dataset_id, None, sources=[store, origin, InstalledConnectors()], store=store) + + assert [(step.release, step.action) for step in plan.steps] == [ + (DEMO, "copy"), + *((reference, "build") for reference in pins), + (CHILD, "read"), + ] + + +@pytest.mark.parametrize( + "parent", [PARENT, DatasetRef.parse("a/parent@1.0.0")], ids=["pinned-build-first", "unpinned-build-first"] +) +def test_two_builds_of_one_release_merge_into_one_build_with_the_pins_of_each(tmp_path, monkeypatch, parent): + # CHILD reads the parent and OTHER, and the card of the parent declares OTHER. So the build of + # the parent builds OTHER without pins, and the read builds OTHER with the pins of CHILD. The read + # plans the payload sources in the order of their names. + cards = [_card(parent, parents={ParentAlias("other"): OTHER}), _card(OTHER)] + store = LocalRegistry(tmp_path / "store") + pins = _child_reading(store, *cards) + calls = _install_demo_backend(monkeypatch, *cards) + sources = [store, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + # The one build of OTHER comes before the build of the parent, which needs it. + assert plan.steps == ( + _build(OTHER, pinned_by={CHILD: pins[OTHER]}), + _build(parent, depends_on=(OTHER,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert calls == [(OTHER, False, {CHILD: pins[OTHER]}), (parent, False, {CHILD: pins[parent]})] + assert Path(version.payload_sources[SourceKey(str(OTHER))].root).is_relative_to(store.root) + + +def test_two_readers_that_pin_one_file_differently_fail_the_read_before_any_step(tmp_path, monkeypatch): + # CHILD reads PARENT and OTHER. The card of PARENT declares X, which the store holds, and which + # reads OTHER too. So the read builds OTHER with the pins of CHILD, and the build of PARENT builds + # OTHER with the pins of X. X pins one file of OTHER with another checksum, so no build matches both. + x = DatasetRef.parse("test/x@1.0.0") + cards = [_card(PARENT, parents={ParentAlias("x"): x}), _card(OTHER)] + store = LocalRegistry(tmp_path / "store") + _child_reading(store, cards[1], child_release=x) + _child_reading(store, *cards) + manifest = store.get_manifest(x.dataset_id) + ((key, pinned),) = manifest.payload_sources.items() + changed = pinned.files[0].model_copy(update={"checksum": "sha256:" + "0" * 64}) + pinned = pinned.model_copy(update={"files": (changed, *pinned.files[1:])}) + (release_dir(store.root, x) / "manifest.json").write_text( + manifest.model_copy(update={"payload_sources": {key: pinned}}).to_json() + ) + # A new registry, so that no manifest that it read before stays. + store = LocalRegistry(store.root) + calls = _install_demo_backend(monkeypatch, *cards) + message = f"{CHILD} and {x} pin {changed.path!r} of {OTHER} as different files" + with pytest.raises(TimeFFormatError, match=f"^{re.escape(message)}"): + read_release(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) + assert calls == [] + + +def test_a_remote_read_fetches_each_manifest_once(tmp_path, monkeypatch): + origin, _ = _child_of_the_demo(tmp_path / "origin") + transport, _ = build_service_fake(origin.root) + remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") + reads = _count_reads(monkeypatch, remote) + + version = read_release(CHILD.dataset_id, None, sources=[remote], store=None) + + assert version.manifest == origin.get_manifest(CHILD.dataset_id) + assert version.payload_sources[SourceKey(str(DEMO))].manifest == origin.get_manifest(DEMO.dataset_id) + # The registry answers "latest" with the manifest, and the plan and the open use that manifest. + assert reads == {f"{CHILD.dataset_id}@None": 1, str(DEMO): 1} + + +def test_latest_reads_the_manifest_of_the_version_that_the_answer_found(tmp_path, monkeypatch): + origin = LocalRegistry(tmp_path / "origin") + origin.put_dataset(make_dataset()) + manifest = origin.get_manifest(DEMO.dataset_id, str(DEMO.version)) + reads = _count_reads(monkeypatch, origin) + # A local registry answers "latest" without its manifest, so the plan reads it at the exact version. + version = read_release(DEMO.dataset_id, None, sources=[origin], store=None) + assert version.manifest == manifest + assert reads == {str(DEMO): 1} + + +@pytest.mark.parametrize("parent", [PARENT, DatasetRef.parse("x/parent@1.0.0")], ids=["build-first", "read-first"]) +def test_a_copy_that_a_build_needs_repairs_a_damaged_copy_that_the_read_found_first(tmp_path, monkeypatch, parent): + # CHILD reads the parent and DEMO. The store comes first in the list and holds DEMO damaged, and origin + # holds DEMO. Only the connectors build the parent, and its card declares DEMO. So the build copies DEMO + # from origin over the damaged copy, and the read takes it there. + parent_card = _card(parent, parents={ParentAlias("demo"): DEMO}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + pins = _child_reading(store, parent_card, make_dataset().metadata) + for registry in (store, origin): + registry.put_dataset(make_dataset()) + (release_dir(store.root, DEMO) / "control.duckdb").write_bytes(b"cut short") + _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + assert plan.steps == ( + CopyStep(DEMO, origin), + _build(parent, depends_on=(DEMO,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=tuple(pins)), + ) + version = read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert store.damage(DEMO) is None + assert Path(version.payload_sources[SourceKey(str(DEMO))].root).is_relative_to(store.root) + + +_SPEC = TimeSeriesSpec(spec_type="sine", name="Sine", unit_value=ureg.dimensionless) + + +def _store_reading(registry: LocalRegistry, release: DatasetRef, source: DatasetRef) -> None: + """Store ``release``, which keeps one Record of ``source`` and adds one Record of its own values.""" + dataset = TimeFDataset(metadata=_card(release, parents={ParentAlias("source"): source})) + with TimeFReader(open_local_version(release_dir(registry.root, source))) as reader: + dataset.import_from(reader, alias=ParentAlias("source"), records=[make_dataset().records[0].id]) + own = Signal.from_values( + np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16) + ) + dataset.add_record(record=Record(sources=(Source(name="own", signals=(own,)),))) + registry.put_dataset(dataset) + + +def test_a_payload_source_that_a_build_needs_as_a_parent_comes_with_its_own_payload_sources(tmp_path, monkeypatch): + # CHILD reads the own Record of SOURCE, which reads DEMO, and a Record of the parent. Origin holds SOURCE + # and DEMO. Only the connectors build the parent, and its card declares SOURCE. The read finds SOURCE in + # origin first. The build then needs SOURCE in the store, so the copy takes DEMO along. + source, parent = DatasetRef.parse("test/source@1.0.0"), DatasetRef.parse("x/parent@1.0.0") + parent_card = _card(parent, parents={ParentAlias("source"): source}) + store, origin = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "origin") + store.put_dataset(make_dataset()) + _store_reading(store, source, DEMO) + store.put_dataset(_demo(parent_card)) + child = TimeFDataset(metadata=_card(CHILD, parents={ParentAlias("source"): source, ParentAlias("parent"): parent})) + for alias, reference, record in (("source", source, 2), ("parent", parent, make_dataset().records[0].id)): + with TimeFReader(open_local_version(release_dir(store.root, reference))) as reader: + child.import_from(reader, alias=ParentAlias(alias), records=[record]) + store.put_dataset(child) + for reference in (DEMO, source, parent): + shutil.copytree(release_dir(store.root, reference), release_dir(origin.root, reference)) + shutil.rmtree(release_dir(store.root, reference)) + shutil.rmtree(release_dir(origin.root, parent)) + pins = {pinned.release: pinned for pinned in store.get_manifest(CHILD.dataset_id).payload_sources.values()} + assert list(pins) == [source, parent] + _install_demo_backend(monkeypatch, parent_card) + sources = [store, origin, InstalledConnectors()] + + plan = plan_read(CHILD.dataset_id, None, sources=sources, store=store) + + assert plan.steps == ( + CopyStep(source, origin), + _build(parent, depends_on=(source,), pinned_by={CHILD: pins[parent]}), + ReadStep(CHILD, store, depends_on=(source, parent)), + ) + read_release(CHILD.dataset_id, None, sources=sources, store=store) + assert store.exists(DEMO.dataset_id, str(DEMO.version)) + + +def test_a_read_of_the_latest_version_asks_each_source_once(tmp_path, monkeypatch): + # The first registry holds DEMO 1.0.0, and the second holds DEMO 2.0.0. + newer = DatasetRef.parse(f"{DEMO.dataset_id}@2.0.0") + registries = [] + for name, dataset in (("first", make_dataset()), ("second", _demo(_card(newer)))): + LocalRegistry(tmp_path / name).put_dataset(dataset) + transport, _ = build_service_fake(tmp_path / name) + registries.append(RemoteRegistry(f"http://{name}.local", transport=transport, cache_dir=tmp_path / f"{name}-c")) + reads = _count_reads(monkeypatch, registries[0]) + + version = read_release(DEMO.dataset_id, "latest", sources=registries, store=None) + + assert version.manifest.metadata.dataset_version == newer.version + # The plan takes the answer of the sources for the latest version, and asks for 2.0.0 nowhere. + assert reads == {f"{DEMO.dataset_id}@None": 1} + + +def test_a_cycle_through_the_requested_release_starts_at_it(tmp_path, monkeypatch): + # CHILD reads PARENT, which only the connectors build, and the card of PARENT declares CHILD. + parent_card = _card(PARENT, parents={ParentAlias("child"): CHILD}) + store = LocalRegistry(tmp_path / "store") + _child_reading(store, parent_card) + _install_demo_backend(monkeypatch, parent_card) + + message = f"dataset parent cycle: {CHILD} -> {PARENT} -> {CHILD}" + with pytest.raises(TimeFValidationError, match=f"^{re.escape(message)}$"): + plan_read(CHILD.dataset_id, None, sources=[store, InstalledConnectors()], store=store) diff --git a/packages/timenet/tests/reader/test_shared_storage.py b/packages/timenet/tests/reader/test_shared_storage.py index 55730194a..0d9a7b49b 100644 --- a/packages/timenet/tests/reader/test_shared_storage.py +++ b/packages/timenet/tests/reader/test_shared_storage.py @@ -394,8 +394,6 @@ def test_opening_a_child_detects_a_forced_replace_of_its_source(tmp_path): message = rf"payload source test/p@1\.0\.0 differs from the release test/c@1\.0\.0 was written against: test/c@1\.0\.0 pins '{first}' as \d+ bytes" with pytest.raises(TimeFFormatError, match=message): open_local_version(child_dir) - with pytest.raises(TimeFFormatError, match=message): - LocalRegistry(tmp_path).open_version("test/c", "1.0.0") def test_storing_a_child_detects_a_forced_replace_of_its_source(tmp_path): @@ -420,7 +418,7 @@ def test_a_child_is_refused_by_a_registry_that_holds_another_release_of_its_sour ours, foreign = LocalRegistry(tmp_path / "ours"), LocalRegistry(tmp_path / "foreign") ours.put_dataset(_release("test/p", [_record_in_memory("p", "Ours. " * 300, [0, 1_000])])) foreign.put_dataset(_release("test/p", [_record_in_memory("p", "Foreign. " * 300, [0, 3_000])])) - with TimeFReader(ours.open_version("test/p", "1.0.0")) as reader: + with TimeFReader(open_local_version(ours.root / "test/p/1.0.0")) as reader: loaded = reader.read() with pytest.raises( TimeFFormatError, diff --git a/packages/timenet/tests/registry/conftest.py b/packages/timenet/tests/registry/conftest.py index 4202abcb6..2b736d6c5 100644 --- a/packages/timenet/tests/registry/conftest.py +++ b/packages/timenet/tests/registry/conftest.py @@ -1,16 +1,8 @@ from pathlib import Path -import sys import pytest -from timenet.dataset import Record, Source - - -# Make sibling test helpers (e.g. ``_fake_registry``) importable by name under -# pytest's importlib import mode, which does not add test dirs to ``sys.path``. -sys.path.insert(0, str(Path(__file__).parent)) - -from timenet.dataset import Signal, TimeFDataset +from timenet.dataset import Record, Signal, Source, TimeFDataset from timenet.dataset.axis import RegularAxis from timenet.testing import make_dataset, sine_loader from timenet.types import ( diff --git a/packages/timenet/tests/registry/test_conformance.py b/packages/timenet/tests/registry/test_conformance.py index e22356e5d..a02a657a3 100644 --- a/packages/timenet/tests/registry/test_conformance.py +++ b/packages/timenet/tests/registry/test_conformance.py @@ -9,6 +9,7 @@ from typing import NamedTuple import uuid +from _demo_child import _child_of_the_demo from _fake_registry import build_service_fake import pytest @@ -16,6 +17,7 @@ from timenet.dataset.axis import RegularAxis from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError from timenet.registry import LocalRegistry, RemoteRegistry, S3Registry, WritableRegistry +from timenet.registry._paths import release_dir from timenet.sources import FoundInRegistry, NotFound from timenet.testing import make_dataset, sine_loader from timenet.types import ( @@ -184,13 +186,34 @@ def test_open_file_rejects_traversal(backend): backend.registry.open_file("timenet/hello-world", "1.0.0", "../../escape") -# ---- open_version ----------------------------------------------------------------------------- +# ---- open_release ----------------------------------------------------------------------------- -def test_open_version_manifest_matches(backend): - version = backend.registry.open_version("timenet/hello-world") # no pin -> latest - assert version.manifest.metadata.dataset_id == "timenet/hello-world" - assert str(version.manifest.metadata.dataset_version) == "1.1.0" +def test_open_release_opens_one_release_without_its_payload_sources(backend, tmp_path): + source, _ = _child_of_the_demo(tmp_path / "source") + child = DatasetRef.parse("test/child@1.0.0") + manifest = source.get_manifest(child.dataset_id) + backend.registry.put_release(release_dir(source.root, child)) + version = backend.registry.open_release(child) + # The child reads files of the demo release, but the handle opens none of them. + assert manifest.payload_sources + assert version.payload_sources == {} + assert version.manifest == backend.registry.get_manifest(child.dataset_id, str(child.version)) + + +def test_open_release_uses_the_manifest_that_the_caller_passes(backend, tmp_path, monkeypatch): + source, _ = _child_of_the_demo(tmp_path / "source") + child = DatasetRef.parse("test/child@1.0.0") + manifest = source.get_manifest(child.dataset_id) + backend.registry.put_release(release_dir(source.root, child)) + + def no_read(*args, **kwargs): + raise AssertionError("the registry reads the manifest that the caller passed") + + monkeypatch.setattr(backend.registry, "get_manifest", no_read) + # No cache holds a copy of the child, so S3 and the remote registry open their remote files. + version = backend.registry.open_release(child, manifest=manifest) + assert version.manifest == manifest # ---- exists ----------------------------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py index 01e1438f4..7459a30ba 100644 --- a/packages/timenet/tests/registry/test_registry.py +++ b/packages/timenet/tests/registry/test_registry.py @@ -4,13 +4,12 @@ from pydantic import ValidationError import pytest -from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetRegistryError +from timenet.errors import TimeFFormatError, TimeNetFileError, TimeNetRegistryError from timenet.manifest import Manifest from timenet.reader import TimeFReader from timenet.registry import ( TIMENET_REGISTRY_URL, BaseRegistry, - DatasetVersion, LocalRegistry, RemoteRegistry, S3Registry, @@ -182,36 +181,10 @@ def test_latest_version_ignores_staging_dirs(registry_root): assert manifest.metadata.dataset_version.major == 2 -# ---- open_version ----------------------------------------------------------------------------- - - -def test_open_version_returns_handle(registry_root): - version = LocalRegistry(registry_root).open_version("demo/ecg") - assert isinstance(version, DatasetVersion) - assert version.manifest.metadata.dataset_id == "demo/ecg" - assert version.root == str(registry_root / "demo/ecg" / "2.0.0") - - -def test_open_version_resolves_latest_to_a_concrete_directory(registry_root): - # ``None`` means latest, but the handle roots at that resolved version, never the string "latest" - version = LocalRegistry(registry_root).open_version("demo/ecg", version=None) - assert version.root.endswith("/2.0.0") - - -def test_open_version_filesystem_reads_a_file(registry_root): - version = LocalRegistry(registry_root).open_version("demo/ecg") - with version.filesystem.open_input_file(version.resolve_file_path("manifest.json")) as handle: - assert b"demo/ecg" in handle.read() - - -def test_open_version_unknown_raises(registry_root): - with pytest.raises(TimeNetDatasetNotFoundError): - LocalRegistry(registry_root).open_version("does/not-exist") - - -def test_open_version_unknown_version_raises(registry_root): - with pytest.raises(TimeNetDatasetNotFoundError): - LocalRegistry(registry_root).open_version("demo/ecg", version="9.9.9") +def test_open_release_checks_the_size_of_each_file(registry_root): + (registry_root / "demo/ecg/2.0.0/control.duckdb").write_bytes(b"cut short") + with pytest.raises(TimeNetFileError, match=r"^demo/ecg@2\.0\.0: file 'control\.duckdb' has 9 bytes"): + LocalRegistry(registry_root).open_release(DatasetRef.parse("demo/ecg@2.0.0")) # ---- search ----------------------------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index c00426a6a..7d8d6ef7c 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -401,7 +401,7 @@ def test_put_dataset_refuses_a_payload_source_the_service_does_not_hold_before_p local.put_dataset(make_dataset()) transport, state = build_publish_fake(token="tok_rw") registry = RemoteRegistry("http://api.local", token="tok_rw", transport=transport, cache_dir=tmp_path / "cache") - with TimeFReader(local.open_version(make_dataset().metadata.dataset_id)) as reader: + with TimeFReader(open_local_version(local.root / "timenet/hello-world/1.0.0")) as reader: parent = reader.read() child = TimeFDataset.from_parts( metadata=parent.metadata.model_copy(update={"dataset_id": "test/child"}), @@ -508,7 +508,7 @@ def test_child_streams_physical_sources_and_downloads_for_offline_reads(tmp_path parent._import_state.parent_cards[ParentAlias("build-only")] = ParentDatasetCard(metadata=build_only) local.put_dataset(parent, values_backend=backend) parent_id = parent.metadata.dataset_id - with TimeFReader(local.open_version(parent_id, "1.0.0")) as reader: + with TimeFReader(open_local_version(local.root / parent_id / "1.0.0")) as reader: reference = DatasetRef(dataset_id=parent_id, version=parent.metadata.dataset_version) child = TimeFDataset( metadata=parent.metadata.model_copy( diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py index 62d5f95f5..96f2f30a2 100644 --- a/packages/timenet/tests/registry/test_s3.py +++ b/packages/timenet/tests/registry/test_s3.py @@ -15,7 +15,7 @@ TimeNetRegistryError, ) from timenet.reader import TimeFReader # ruff: ignore[module-import-not-at-top-of-file] -from timenet.registry import LocalRegistry, S3Registry # ruff: ignore[module-import-not-at-top-of-file] +from timenet.registry import LocalRegistry, S3Registry, open_local_version # ruff: ignore[module-import-not-at-top-of-file] from timenet.testing import assert_datasets_equal, make_dataset # ruff: ignore[module-import-not-at-top-of-file] from timenet.types import DatasetSchema, Version # ruff: ignore[module-import-not-at-top-of-file] from timenet.types.references import DatasetRef # ruff: ignore[module-import-not-at-top-of-file] @@ -59,7 +59,9 @@ def test_put_dataset_and_remote_load_round_trip(s3_root, tmp_path, backend, entr assert_datasets_equal(make_dataset(), reader.read()) else: assert_datasets_equal(make_dataset(), client.load(make_dataset().metadata.dataset_id)) - assert isinstance(registry.open_version(make_dataset().metadata.dataset_id).filesystem, pafs.S3FileSystem) + assert isinstance( + registry.open_release(DatasetRef.parse("timenet/hello-world@1.0.0")).filesystem, pafs.S3FileSystem + ) assert not (tmp_path / "cache" / make_dataset().metadata.dataset_id).exists() @@ -68,7 +70,7 @@ def test_put_dataset_checks_payload_sources_against_the_bucket(s3_root, tmp_path local.put_dataset(make_dataset()) # moto keeps one bucket state per process, so this test uses a prefix no other test writes. registry = S3Registry("s3://tn-test/payload-sources", cache_dir=tmp_path / "cache") - with TimeFReader(local.open_version(make_dataset().metadata.dataset_id)) as reader: + with TimeFReader(open_local_version(local.root / "timenet/hello-world/1.0.0")) as reader: parent = reader.read() child = TimeFDataset.from_parts( metadata=parent.metadata.model_copy(update={"dataset_id": "test/child"}), @@ -104,7 +106,7 @@ def test_download_then_load_serves_from_cache(s3_root, tmp_path): client = TimeNet(registry=registry, storage_path=tmp_path / "cache") path = client.download(dataset_id) # boto3 GetObject per file, into the cache assert (path / "manifest.json").exists() - handle = registry.open_version(dataset_id) + handle = registry.open_release(DatasetRef.parse(f"{dataset_id}@1.0.0")) assert isinstance(handle.filesystem, pafs.LocalFileSystem) # cache-first: no S3 reads assert_datasets_equal(make_dataset(), client.load(dataset_id)) diff --git a/packages/timenet/tests/registry/test_version.py b/packages/timenet/tests/registry/test_version.py index d8e235e24..d0e5c3b4b 100644 --- a/packages/timenet/tests/registry/test_version.py +++ b/packages/timenet/tests/registry/test_version.py @@ -10,7 +10,7 @@ from timenet.format.checksums import file_checksum from timenet.manifest import PayloadSource from timenet.registry import LocalRegistry, open_local_version -from timenet.registry.dependencies import open_cached_or_remote, prepare_local_copy +from timenet.registry.dependencies import prepare_local_copy from timenet.registry.shared import complete_copy from timenet.types.reference_keys import SourceKey from timenet.types.references import DatasetRef @@ -153,14 +153,6 @@ def test_preparation_needs_only_the_direct_payload_sources(registry_root, tmp_pa with pytest.raises(TimeNetDatasetNotFoundError, match=rf"payload source {parent_ref} of {reference} is not in"): prepare_local_copy(cache, reference) - def missing_remote(ref): - raise TimeNetDatasetNotFoundError(f"no remote release {ref}") - - with pytest.raises( - TimeNetDatasetNotFoundError, match=rf"payload source {parent_ref} .* is not in remote or its cache" - ): - open_cached_or_remote(cache, reference, missing_remote, searched="remote") - prepare_local_copy(cache, reference, fetch_from=LocalRegistry(registry_root)) cached_parent = cache / "timenet/hello-world/1.0.0" assert (cached_parent / "control.duckdb").is_file() @@ -175,7 +167,6 @@ def missing_remote(ref): prepared = prepare_local_copy(cache, reference) assert set(prepared.payload_sources) == {SourceKey(str(parent_ref))} assert prepared.resolve_payload_source(SourceKey(str(parent_ref))).payload_sources == {} - assert open_cached_or_remote(cache, reference, missing_remote, searched="remote").payload_sources @pytest.mark.parametrize("suffix", ["different/ecg/2.0.0", "demo/ecg/3.0.0", "demo/ecg/latest"]) diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index d75e84864..5c94a5523 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -1,7 +1,7 @@ -from pathlib import Path import shutil from types import SimpleNamespace +from _demo_child import _child_of_the_demo import httpx from pydantic import ValidationError import pytest @@ -35,27 +35,6 @@ from timenet.writer import TimeFWriter -def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: - """Store the demo dataset and ``test/child``, which reads one of its Records, in a local registry. - - Returns: - The registry, and the demo release, which is the payload source of the child. - """ - parent = make_dataset() - registry = LocalRegistry(root) - registry.put_dataset(parent) - reference = DatasetRef(dataset_id=parent.metadata.dataset_id, version=parent.metadata.dataset_version) - child = TimeFDataset( - metadata=parent.metadata.model_copy( - update={"dataset_id": "test/child", "parents": {ParentAlias("parent"): reference}} - ) - ) - with TimeNet(registry).open(parent.metadata.dataset_id) as reader: - child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) - registry.put_dataset(child) - return registry, reference - - class _Builder: """A builder backend whose cards declare ``VERSION``. @@ -441,7 +420,7 @@ def counting_close(reader): def test_load_reads_in_place_without_downloading(registry_root, tmp_path): - # load() reads through the registry's open_version handle, not download-then-read, so it must not + # load() reads the release where it is, not download-then-read, so it must not # write a copy into local storage the way download() does. storage = tmp_path / "store" client = TimeNet(registry_root, storage_path=storage) diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 0dd8fdee5..431b73b86 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -11,7 +11,8 @@ from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError from timenet.manifest import Manifest from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry +from timenet.registry import LocalRegistry, open_local_version +from timenet.registry._paths import release_dir from timenet.testing import make_dataset from timenet.types import Access, Version from timenet.types.access import effective_access @@ -396,8 +397,8 @@ def test_a_build_context_made_directly_closes_its_parent_readers(tmp_path, monke for dataset in (make_dataset(), second): registry.put_dataset(dataset) readers = { - ParentAlias("parent"): TimeFReader(registry.open_version(_PARENT.dataset_id, str(_PARENT.version))), - ParentAlias("second"): TimeFReader(registry.open_version("test/second", "1.0.0")), + ParentAlias("parent"): TimeFReader(open_local_version(release_dir(registry.root, _PARENT))), + ParentAlias("second"): TimeFReader(open_local_version(registry.root / "test/second/1.0.0")), } closed = [] close = TimeFReader.close