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 52e40c3d3..53e01c97e 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 @@ -187,7 +187,7 @@ def test_equivalent_series_encodings_share_a_split(series, tmp_path): def test_splits_survive_storage_round_trip(tmp_path): original = TSQAConnector().convert(_refs(tmp_path, _balanced_rows())) registry = LocalRegistry(tmp_path / "registry") - registry.store(original) + registry.put_dataset(original) with TimeFReader(registry.open_version("chengsenwang/tsqa", "1.2.0")) as reader: restored = reader.read() assert {read_value(task.prompt): task.split for task in restored.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 b158657fb..93f4c09ba 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 @@ -125,7 +125,7 @@ def test_every_task_holds_its_published_row(): def test_storage_round_trip_keeps_the_whole_conversion(tmp_path): original = _FixtureConnector().convert(_refs()) registry = LocalRegistry(tmp_path / "registry") - registry.store(original) + registry.put_dataset(original) published = _published() with TimeFReader(registry.open_version("tsaqa/tsaqa-benchmark", "1.0.0")) as reader: restored = reader.read() diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index 5da7d7c06..1747b5722 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -120,7 +120,7 @@ def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_pa _install_cards(monkeypatch, child={"parent": PARENT, "grand": GRAND}, parent={"grand": GRAND}) registries = {name: LocalRegistry(tmp_path / name) for name in ("output", "origin")} if grand_in in registries: - registries[grand_in].store(make_dataset()) + registries[grand_in].put_dataset(make_dataset()) steps = plan_parents(CHILD.dataset_id, registries["output"], fetch_from=registries["origin"], build_parents=True) @@ -182,7 +182,7 @@ def _child_of(registry: LocalRegistry, reference: DatasetRef, parent: DatasetRef np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16) ) child.add_record(record=Record(sources=(Source(name="own", signals=(own,)),))) - registry.store(child) + registry.put_dataset(child) def _declared_without_its_sources_source(registry: LocalRegistry) -> None: @@ -190,7 +190,7 @@ def _declared_without_its_sources_source(registry: LocalRegistry) -> None: DECLARED's payload sources are SOURCE alone; SOURCE's are GRAND. GRAND is removed afterwards. """ - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) _child_of(registry, SOURCE, GRAND, make_dataset().records[0].id) # SOURCE's own Record, which holds values it stores itself, takes the second Record ID. _child_of(registry, DECLARED, SOURCE, 2) @@ -279,7 +279,7 @@ def convert(self, raw_refs): def _parent_without_its_source(output: LocalRegistry) -> None: """Store GRAND and PARENT, which reads a GRAND Record, then remove GRAND.""" - output.store(make_dataset()) + output.put_dataset(make_dataset()) _child_of(output, PARENT, GRAND, make_dataset().records[0].id) shutil.rmtree(release_dir(output.root, GRAND)) @@ -326,7 +326,7 @@ def test_a_payload_source_that_two_releases_read_is_built_against_the_pins_of_bo sibling = DatasetRef.parse("test/sibling@1.0.0") _install_cards(monkeypatch, child={"parent": PARENT, "sibling": sibling}) output = LocalRegistry(tmp_path / "output") - output.store(make_dataset()) + output.put_dataset(make_dataset()) # PARENT and the sibling each read another Record of GRAND, so their pins differ. _child_of(output, PARENT, GRAND, make_dataset().records[0].id) _child_of(output, sibling, GRAND, make_dataset().records[1].id) @@ -357,7 +357,7 @@ def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): assert plan_parents(PARENT.dataset_id, output, build_parents=True, force_build=True) == [ParentStep(GRAND, "build")] # A registry that holds the source fills the gap: the local PARENT is reused and GRAND is copied. origin = LocalRegistry(tmp_path / "origin") - origin.store(make_dataset()) + origin.put_dataset(make_dataset()) assert plan_parents(PARENT.dataset_id, output, fetch_from=origin) == [ParentStep(PARENT, "fetch")] prepare_parents(PARENT.dataset_id, output, build=_no_build, fetch_from=origin) assert output.exists(GRAND.dataset_id, str(GRAND.version)) @@ -367,7 +367,7 @@ def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkey _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") for registry in (output, origin): - registry.store(TimeFDataset(metadata=_cards(parent={})[PARENT.dataset_id])) + registry.put_dataset(TimeFDataset(metadata=_cards(parent={})[PARENT.dataset_id])) (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") with pytest.raises(TimeNetFileError, match=r"'control\.duckdb' has 9 bytes, expected [\d,]+ bytes\.$") as caught: @@ -401,8 +401,11 @@ def open_file(self, dataset_id, version, relpath): def open_version(self, dataset_id, version=None): return self._local.open_version(dataset_id, version) - def store(self, dataset, **options): - return self._local.store(dataset, **options) + def put_dataset(self, dataset, **options): + return self._local.put_dataset(dataset, **options) + + def put_release(self, from_directory): + return self._local.put_release(from_directory) _GATED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} @@ -415,7 +418,7 @@ def test_a_remote_output_refuses_a_plan_that_visits_a_release_that_is_not_open(t cards[gated_ref.dataset_id] = cards[gated_ref.dataset_id].model_copy(update=_GATED) output = _Hosted(tmp_path / "hosted") if gated == "held-parent": - output.store(TimeFDataset(metadata=cards[PARENT.dataset_id])) + output.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) with pytest.raises(TimeNetAccessError) as caught: plan_parents(CHILD.dataset_id, output, build_parents=True) @@ -432,7 +435,7 @@ def test_a_remote_output_names_the_access_page_of_a_held_parent_from_its_lineage held = TimeFDataset(metadata=cards[PARENT.dataset_id]) grand = make_dataset().metadata.model_copy(update=_GATED) held._import_state.lineage[str(GRAND)] = ParentDatasetCard(metadata=grand) - output.store(held) + output.put_dataset(held) with pytest.raises(TimeNetAccessError) as caught: plan_parents(CHILD.dataset_id, output, build_parents=True) @@ -458,7 +461,7 @@ def run_isolated(dataset_id, root, *, force=False, **_): dataset._metadata = metadata for handle in context.parents.values(): dataset.import_from(handle, records=[make_dataset().records[0].id]) - registry.store(dataset, force=force) + registry.put_dataset(dataset, force=force) return str(registry.root / dataset_id / str(metadata.dataset_version)) return run_isolated @@ -469,7 +472,7 @@ def test_build_prepares_parents_before_the_isolated_child(tmp_path, monkeypatch) monkeypatch.setattr(cli_module, "connector_dir", lambda dataset_id: tmp_path) monkeypatch.delenv("TIMENET_ISOLATION", raising=False) origin = LocalRegistry(tmp_path / "origin") - origin.store(make_dataset()) + origin.put_dataset(make_dataset()) out = tmp_path / "out" calls = [] monkeypatch.setattr(cli_module, "run_isolated", _isolated_builds(calls)) @@ -490,7 +493,7 @@ def test_a_built_version_starts_no_isolated_child(tmp_path, monkeypatch): monkeypatch.delenv("TIMENET_ISOLATION", raising=False) monkeypatch.setattr(env_module.subprocess, "Popen", lambda *args, **kwargs: pytest.fail("started a subprocess")) out = LocalRegistry(tmp_path / "out") - out.store(make_dataset()) + out.put_dataset(make_dataset()) assert cli_module.build_one(GRAND.dataset_id, out.root) == str(release_dir(out.root, GRAND)) @@ -516,7 +519,7 @@ def test_every_build_entry_point_prepares_parents(tmp_path, monkeypatch, entry): origin = LocalRegistry(tmp_path / "origin") parent = make_dataset() parent._metadata = cards[PARENT.dataset_id] - origin.store(parent) + origin.put_dataset(parent) build(CHILD.dataset_id, fetch_from=origin, **{"out" if entry == "api" else "root": out}) assert calls == [(CHILD.dataset_id, False)] assert LocalRegistry(out).exists(PARENT.dataset_id, str(PARENT.version)) @@ -533,7 +536,7 @@ def test_load_obtains_a_missing_parent_before_it_builds_and_reads(tmp_path, monk origin = LocalRegistry(tmp_path / "origin") parent = make_dataset() parent._metadata = cards[PARENT.dataset_id] - origin.store(parent) + origin.put_dataset(parent) with pytest.raises(TimeNetMissingParentError, match=r"^parent release test/parent@1\.0\.0 of test/child"): timenet_connectors.load(CHILD.dataset_id) @@ -553,7 +556,7 @@ def test_load_copies_a_missing_payload_source_through_fetch_from(tmp_path, monke out = LocalRegistry(tmp_path / "out") _parent_without_its_source(out) origin = LocalRegistry(tmp_path / "origin") - origin.store(make_dataset()) + origin.put_dataset(make_dataset()) monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) monkeypatch.setenv("TIMENET_REGISTRY", str(out.root)) # The build has nothing to do, and no builder backend can build the payload source, so only the @@ -579,7 +582,7 @@ def test_a_fetch_of_a_release_that_is_not_open_fails_before_any_download(tmp_pat cards = _install_cards(monkeypatch, child={"grand": GRAND, "parent": PARENT}, parent={}) origin = _NoDownloads(tmp_path / "origin") # GRAND is open and would be fetched first. - origin.store(make_dataset()) + origin.put_dataset(make_dataset()) ancestor = DatasetRef.parse("test/ancestor@1.0.0") if page == "own": parent, requirement = TimeFDataset(metadata=cards[PARENT.dataset_id].model_copy(update=_GATED)), PARENT @@ -589,7 +592,7 @@ def test_a_fetch_of_a_release_that_is_not_open_fails_before_any_download(tmp_pat card = make_dataset().metadata.model_copy(update={"dataset_id": ancestor.dataset_id, **_GATED}) parent._import_state.lineage[str(ancestor)] = ParentDatasetCard(metadata=card) requirement = ancestor - origin.store(parent) + origin.put_dataset(parent) output = LocalRegistry(tmp_path / "out") with pytest.raises(TimeNetAccessError) as caught: @@ -610,7 +613,7 @@ def test_a_pinned_parent_build_hands_its_pins_to_the_isolated_child(tmp_path, mo def run_pipeline(connector, root, *, pinned_by=None, **_): received.append(pinned_by) if pinned_by: - output.store(make_dataset()) + output.put_dataset(make_dataset()) return Path(root) def child(command, env): @@ -658,7 +661,7 @@ def test_a_parent_that_another_backend_claims_builds_there_with_its_pins(tmp_pat output = LocalRegistry(tmp_path / "output") stored = make_dataset() stored._metadata = stored.metadata.model_copy(update={"dataset_id": STORED.dataset_id}) - output.store(stored) + output.put_dataset(stored) _child_of(output, PARENT, STORED, make_dataset().records[0].id) pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(STORED))] saved = tmp_path / "saved" @@ -696,7 +699,7 @@ def build(self, dataset_id, root, **options): self.calls.append((dataset_id, options)) dataset = make_dataset() dataset._metadata = self.cards[dataset_id] - LocalRegistry(root).store(dataset) + LocalRegistry(root).put_dataset(dataset) return release_dir(root, DatasetRef.parse(f"{dataset_id}@{dataset.metadata.dataset_version}")) @@ -758,7 +761,7 @@ def test_the_cli_hint_names_its_flags(tmp_path, monkeypatch, case): monkeypatch.setattr(cli_module, "connector_dir", lambda dataset_id: tmp_path) out = LocalRegistry(tmp_path / "out") if case == "damaged": - out.store(TimeFDataset(metadata=cards[PARENT.dataset_id])) + out.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) (release_dir(out.root, PARENT) / "control.duckdb").write_bytes(b"cut short") if case == "not-open": cards[CHILD.dataset_id] = cards[CHILD.dataset_id].model_copy(update=_GATED) diff --git a/packages/timenet-connectors/tests/test_end_to_end.py b/packages/timenet-connectors/tests/test_end_to_end.py index 78fcf6dd8..702654e58 100644 --- a/packages/timenet-connectors/tests/test_end_to_end.py +++ b/packages/timenet-connectors/tests/test_end_to_end.py @@ -125,7 +125,7 @@ def test_resolve_target_publishes_to_remote_timenet_registry(clean_env, monkeypa def test_publish_pipeline_stores_through_writable_registry(clean_env, converted, tmp_path): - # publish_pipeline hands the converted dataset to registry.store; a LocalRegistry stands in for a + # publish_pipeline hands the converted dataset to registry.put_dataset; a LocalRegistry stands in for a # remote one, so the publish path is exercised without a network round-trip. registry = LocalRegistry(tmp_path / "registry") assert publish_pipeline(HelloWorldConnector(), registry) == "1.0.0" diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py new file mode 100644 index 000000000..ec216e113 --- /dev/null +++ b/packages/timenet/src/timenet/_fs.py @@ -0,0 +1,39 @@ +"""Replace a directory with a complete staging directory.""" + +from pathlib import Path +import shutil +import uuid + + +def publish_directory(staging: Path, final: Path) -> None: + """Rename ``staging`` to ``final``, replacing any existing directory. + + If publishing fails, this function tries to restore the old directory. After a hard kill during + replacement, rename the sibling ``.old-`` directory back to ``final`` to recover it. + + Args: + staging: The completed directory, with the same parent as ``final``. + final: The destination directory, whose parent must exist. + + Raises: + OSError: If a rename fails, with the backup path if the old directory cannot be restored. + """ + old = None + if final.exists(): + old = final.with_name(f"{final.name}.old-{uuid.uuid4().hex}") + final.replace(old) + try: + staging.replace(final) + except BaseException as error: + if old is not None: + try: + old.replace(final) + except OSError as restore_error: + raise OSError( + f"could not publish {final} ({error}) or move the old release back " + f"({restore_error}); it is in {old}, so rename that to {final} to recover it" + ) from error + raise + if old is not None: + # The new directory is published, so failing to delete the old one must not fail the publish. + shutil.rmtree(old, ignore_errors=True) diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index c7a392b08..d592bf0d4 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -113,26 +113,26 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] force: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Build and publish a dataset to a writable registry. + """Run a connector and publish its dataset to a registry. - The build reuses an existing version unless ``force`` is true. - All declared parents must already exist in the registry. + If the version already exists, return it unless ``force`` is set. Remote and S3 registries + require ``access: open`` on the dataset card and its lineage. Args: connector: The connector to build. - registry: The local, remote, or S3 registry to publish to. - cache_dir: Download directory to keep, or ``None`` for ``/``. - keep_cache: Keep the default download cache after publishing. - progress_cb: Optional writer progress callback. - force: Republish an existing version. - pinned_by: Releases that read this dataset's files, with file pins the build must match. + registry: The destination registry. + cache_dir: Download directory, or ``None`` for ``/``. + keep_cache: Keep the default cache after publishing (a supplied ``cache_dir`` is always kept). + progress_cb: Optional callback for write progress. + force: Republish even if the version already exists. + pinned_by: Releases that already read this release's files, with the pins that the new files must match. Returns: The published version string. Raises: - TimeFValidationError: If a card with parents has no ``compose`` override, or ``compose`` changes the card. - TimeNetAccessError: If a release or ancestor lacks open access for a nonlocal registry. + TimeFValidationError: If declared parents lack a custom ``compose``, or ``compose`` changes the dataset card. + TimeNetAccessError: If a nonlocal publish includes a dataset or ancestor without open access. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access values_backend = _check_values_backend(connector) metadata = connector.metadata() @@ -152,10 +152,9 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] if not isinstance(registry, LocalRegistry): require_open_access(dataset.metadata, dataset.lineage.values()) dataset.derive_schema() - # Pass pins only when there are any, so a registry written against the earlier store() - # signature keeps working for every other build. - pinned = {"pinned_by": pinned_by} if pinned_by else {} - registry.store(dataset, force=force, values_backend=values_backend, progress_cb=progress_cb, **pinned) + registry.put_dataset( + dataset, force=force, values_backend=values_backend, progress_cb=progress_cb, pinned_by=pinned_by + ) if not keep_cache and cache_dir is None and cache.is_dir(): shutil.rmtree(cache) return version diff --git a/packages/timenet/src/timenet/manifest/__init__.py b/packages/timenet/src/timenet/manifest/__init__.py index bcf402448..480bdf788 100644 --- a/packages/timenet/src/timenet/manifest/__init__.py +++ b/packages/timenet/src/timenet/manifest/__init__.py @@ -2,8 +2,17 @@ from timenet.manifest.counts import ManifestCounts from timenet.manifest.files import FilePart, ManifestFiles -from timenet.manifest.manifest import BuildEnvironment, Manifest +from timenet.manifest.manifest import BuildEnvironment, Manifest, parse_manifest, read_manifest from timenet.manifest.payload_source import PayloadSource -__all__ = ["BuildEnvironment", "FilePart", "Manifest", "ManifestCounts", "ManifestFiles", "PayloadSource"] +__all__ = [ + "BuildEnvironment", + "FilePart", + "Manifest", + "ManifestCounts", + "ManifestFiles", + "PayloadSource", + "parse_manifest", + "read_manifest", +] diff --git a/packages/timenet/src/timenet/manifest/manifest.py b/packages/timenet/src/timenet/manifest/manifest.py index bfa678fc7..f93815e71 100644 --- a/packages/timenet/src/timenet/manifest/manifest.py +++ b/packages/timenet/src/timenet/manifest/manifest.py @@ -1,10 +1,12 @@ """The :class:`Manifest`: the compiled ``manifest.json`` file and its JSON codec.""" +from pathlib import Path from typing import Annotated, Any, Self -from pydantic import Field, StrictStr, model_validator +from pydantic import Field, StrictStr, ValidationError, model_validator from typing_extensions import TypedDict +from timenet.errors import TimeFFormatError from timenet.manifest.counts import ManifestCounts from timenet.manifest.files import ManifestFiles from timenet.manifest.payload_source import PayloadSource @@ -102,3 +104,45 @@ def to_json(self) -> str: # Apply the same validation as to_dict() to catch changes in nested dictionaries. validated = type(self).model_validate(self) return validated.model_dump_json(indent=2, by_alias=True) + + +def parse_manifest(text: str | bytes, where: str) -> Manifest: + """Parse the JSON text of a stored manifest. + + Args: + text: The JSON text. + where: The source location to include in errors. + + Returns: + The manifest. + + Raises: + TimeFFormatError: If the text is not a valid manifest. + """ + try: + return Manifest.model_validate_json(text) + except ValidationError as error: + first = error.errors()[0] + location = ".".join(str(part) for part in first["loc"]) + message = first["msg"].removeprefix("Value error, ") + detail = f"{location}: {message}" if location else message + raise TimeFFormatError(f"{where} is not a valid manifest: {detail}") from error + + +def read_manifest(path: Path) -> Manifest: + """Read a ``manifest.json`` file. + + Args: + path: The file. + + Returns: + The manifest. + + Raises: + TimeFFormatError: If the file cannot be read or is not a valid manifest. + """ + try: + text = path.read_bytes() + except OSError as error: + raise TimeFFormatError(f"cannot read the manifest {path}: {error.strerror or error}") from error + return parse_manifest(text, str(path)) diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index ce92bc9f6..8836975e7 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -23,7 +23,7 @@ TimeNetMissingParentError, ) from timenet.format.constants import MANIFEST_FILE -from timenet.manifest import FilePart, Manifest, PayloadSource +from timenet.manifest import FilePart, Manifest, PayloadSource, read_manifest from timenet.registry._paths import release_dir from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion @@ -63,7 +63,7 @@ def open_local_version(path: str | Path, *, sources_root: str | Path | None = No manifest_path = directory / MANIFEST_FILE if not manifest_path.is_file(): raise FileNotFoundError(f"no manifest at {manifest_path}") - manifest = Manifest.model_validate_json(manifest_path.read_text()) + manifest = read_manifest(manifest_path) if not manifest.payload_sources: return open_local_release(directory, manifest, sources_root=None) if sources_root is None: @@ -77,15 +77,13 @@ def open_local_release(directory: Path, manifest: Manifest, *, sources_root: Pat Args: directory: The release directory. manifest: The release's manifest. - sources_root: The registry root that holds its payload sources. Only a release without - payload sources may omit it. + sources_root: The registry root of the payload sources, required if the release has any. Returns: A handle with each payload source opened and checked. Raises: - TimeNetFileError: If a file of the release or of a payload source is missing or has another - size than recorded. + TimeNetFileError: If a required file is missing or its size differs from its manifest record. TimeFFormatError: If a payload source differs from the files the release pins. TimeNetMissingParentError: If a payload source is not under ``sources_root``. TimeFValidationError: If the release has payload sources but no ``sources_root``. @@ -103,9 +101,7 @@ def open_source(reference: DatasetRef) -> DatasetVersion: source_manifest = source_dir / MANIFEST_FILE if not source_manifest.is_file(): raise TimeNetDatasetNotFoundError(f"no manifest at {source_manifest}") - return DatasetVersion( - Manifest.model_validate_json(source_manifest.read_text()), version.filesystem, str(source_dir) - ) + return DatasetVersion(read_manifest(source_manifest), version.filesystem, str(source_dir)) return attach_payload_sources(version, open_source, searched=str(root)) @@ -311,17 +307,13 @@ def _ensure_dataset_version_available_locally( # ruff: ignore[too-many-argument Raises: TimeNetDatasetNotFoundError: If the release is absent and this function cannot fetch it. - TimeNetFileError: If the local manifest is present but a file is missing or has another - size, and ``fetch_from`` is not given. Also if a fetched file does not match its - manifest record. + TimeNetFileError: If a local file is damaged and ``fetch_from`` is not given, or a fetched file does not match. TimeFFormatError: If a manifest identifies a different release. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers if manifest is not None: validate_manifest_reference(manifest, reference) if not force and (directory / MANIFEST_FILE).is_file(): - local = ( - manifest if manifest is not None else Manifest.model_validate_json((directory / MANIFEST_FILE).read_text()) - ) + local = manifest if manifest is not None else read_manifest(directory / MANIFEST_FILE) validate_manifest_reference(local, reference) try: check_file_sizes( @@ -350,14 +342,13 @@ def _load_complete_local_manifest( """Load and validate a local release, returning None if files are missing or incomplete. Returns: - The manifest if all declared files exist with their expected sizes, otherwise None. - Invalid manifests raise an error instead of being treated as missing releases. + The manifest if it exists and every listed file has the recorded size, otherwise ``None``. """ manifest_path = directory / MANIFEST_FILE if not manifest_path.is_file(): return None if manifest is None: - manifest = Manifest.model_validate_json(manifest_path.read_text()) + manifest = read_manifest(manifest_path) validate_manifest_reference(manifest, reference) return manifest if complete_copy(directory, manifest) else None diff --git a/packages/timenet/src/timenet/registry/factory.py b/packages/timenet/src/timenet/registry/factory.py index 73d566bc9..506fef9f9 100644 --- a/packages/timenet/src/timenet/registry/factory.py +++ b/packages/timenet/src/timenet/registry/factory.py @@ -62,16 +62,16 @@ def open_registry(uri: str | Path, *, cache_dir: str | Path | None = None) -> Ba def open_writable_registry(uri: str | Path) -> WritableRegistry: - """Open a registry that supports :meth:`~WritableRegistry.store`, for build to publish into. + """Open a writable registry from a URI or local path. Args: uri: A URL, ``timenet://`` / ``s3://`` / ``file://`` URI, or local path. Returns: - The matching writable registry backend. + The writable registry for the URI or path. Raises: - ValueError: If the resolved backend does not support writing. + ValueError: If the registry does not support writing. """ registry = open_registry(uri) if not isinstance(registry, WritableRegistry): diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index 7c6fc323f..0b6c944ac 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -1,11 +1,20 @@ """Shared opening logic for registries that read releases remotely.""" from abc import ABC, abstractmethod +from collections.abc import Callable, Mapping +from pathlib import Path +import shutil +import tempfile +from typing import ClassVar +from timenet.dataset import TimeFDataset +from timenet.manifest import PayloadSource +from timenet.registry._paths import release_dir from timenet.registry.dependencies import open_cached_or_remote from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry from timenet.types.references import DatasetRef, is_latest +from timenet.writer import TimeFWriter, WriteProgressEvent class LazyReadRegistry(WritableRegistry, ABC): @@ -15,6 +24,55 @@ class LazyReadRegistry(WritableRegistry, ABC): :meth:`_open_remote_version` is the one method they implement for reading. """ + _staging_prefix: ClassVar[str] = "timenet-publish-" + """The name prefix of the local folder that :meth:`put_dataset` writes into before it publishes.""" + + def put_dataset( + self, + dataset: TimeFDataset, + *, + force: bool = False, + values_backend: str = "parquet", + progress_cb: Callable[[WriteProgressEvent], None] | None = None, + pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, + ) -> DatasetRef: + """Write a dataset locally and publish it to this registry. + + Return the existing release's reference unless ``force`` is set. + + Args: + dataset: The dataset to write. + force: Publish even if the version already exists. + values_backend: Storage format for values (``"parquet"`` or ``"zarr"``). + progress_cb: Optional callback for write progress. + pinned_by: Releases that already read this release's files, with the pins that the new files must match. + + Returns: + A reference to the stored release. + """ + if dataset.schema is None: + dataset.derive_schema() + dataset_id = dataset.metadata.dataset_id + version = str(dataset.metadata.dataset_version) + reference = DatasetRef.parse(f"{dataset_id}@{version}") + if not force and self.exists(dataset_id, version): + return reference + staging = Path(tempfile.mkdtemp(prefix=self._staging_prefix)) + try: + # The writer stages locally but checks every payload source against this registry. + with TimeFWriter( + staging, + dataset, + values_backend=values_backend, + progress_cb=progress_cb, + registry=self, + pinned_by=pinned_by, + ) as writer: + writer.write() + return self.put_release(release_dir(staging, reference)) + 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. diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 54b030417..81b3a2eaf 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -2,12 +2,15 @@ from collections.abc import Callable, Mapping from pathlib import Path +import shutil from typing import BinaryIO +import uuid +from timenet._fs import publish_directory from timenet.dataset import TimeFDataset from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError from timenet.format.constants import MANIFEST_FILE -from timenet.manifest import Manifest, PayloadSource +from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry._paths import release_dir from timenet.registry.dependencies import open_local_release from timenet.registry.version import DatasetVersion @@ -19,7 +22,7 @@ class LocalRegistry(WritableRegistry): - """Serve datasets from a local directory. The build output is itself a valid registry.""" + """Read and publish datasets in a local directory.""" def __init__(self, root: Path) -> None: """Open a local registry rooted at a directory. @@ -85,8 +88,7 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: Raises: TimeNetDatasetNotFoundError: If the dataset id or version has no committed manifest. - TimeFFormatError: If the stored manifest declares a dataset id that differs from its - directory. This means a misplaced or corrupt artifact. + TimeFFormatError: If the manifest is unreadable, invalid, or names a different dataset. """ resolved = self._latest_version(dataset_id) if is_latest(version) else version if resolved is None: @@ -102,7 +104,7 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: cached = self._manifest_cache.get((dataset_id, resolved)) if cached is not None and cached[0] == mtime: return cached[1] - manifest = Manifest.model_validate_json(path.read_text()) + manifest = read_manifest(path) if manifest.metadata.dataset_id != dataset_id: raise TimeFFormatError( f"manifest under {dataset_id!r}/{resolved} declares dataset id " @@ -135,7 +137,7 @@ def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: raise ValueError(f"relpath {relpath!r} escapes dataset {dataset_id!r} version {version!r}") return target.open("rb") - def store( + def put_dataset( self, dataset: TimeFDataset, *, @@ -143,40 +145,34 @@ def store( values_backend: str = "parquet", progress_cb: Callable[[WriteProgressEvent], None] | None = None, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, - ) -> str: - """Compile a dataset and write it into this registry's directory tree. + ) -> DatasetRef: + """Write a dataset to this local registry. - Stream the dataset through a :class:`~timenet.writer.TimeFWriter`. The writer stages under - ``.tmp-*`` and publishes with a single atomic rename. It skips an already-committed - version unless the caller sets ``force``, which replaces it only once the new release is - ready, so a failed write keeps the old one. The writer checks every payload source of the - release against this registry, since its readers find them here. + Return an existing release's reference unless ``force`` is set. Reused data must match the + source releases stored in this registry. Args: - dataset: The populated dataset to store. - force: Replace an already-committed version instead of skipping it. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). - progress_cb: Optional writer progress callback. - pinned_by: The releases that already read files of this release, with their pins. The - writer publishes only if it reproduced each pinned file. + dataset: The dataset to write. + force: Replace an existing version. + values_backend: Storage format for values (``"parquet"`` or ``"zarr"``). + progress_cb: Optional callback for write progress. + pinned_by: Releases that already read this release's files, with the pins that the new files must match. Returns: - The stored version string. + A reference to the stored release. Raises: - TimeNetDatasetNotFoundError: If the dataset reuses payloads stored by a release that this - registry does not hold. - TimeFFormatError: If this registry holds such a release in another form, or a release in - ``pinned_by`` pins a file that the write did not reproduce. + TimeNetDatasetNotFoundError: If a release that stores reused data is missing from this registry. + TimeFFormatError: If a source release or the new output fails a file pin check. """ # ruff: ignore[docstring-extraneous-exception] - the writer checks payload sources before it publishes if dataset.schema is None: dataset.derive_schema() version = str(dataset.metadata.dataset_version) dataset_id = dataset.metadata.dataset_id # DatasetRef validates the release before a write can touch its directory. - DatasetRef.parse(f"{dataset_id}@{version}") + reference = DatasetRef.parse(f"{dataset_id}@{version}") if self.exists(dataset_id, version) and not force: - return version + return reference with TimeFWriter( self._root, dataset, @@ -187,7 +183,39 @@ def store( pinned_by=pinned_by, ) as writer: writer.write() - return version + return reference + + def put_release(self, from_directory: Path) -> DatasetRef: + """Copy a release directory into this registry. + + The directory must contain ``manifest.json`` and every file it lists. This method copies + those files before replacing an existing version. + + Args: + from_directory: The release directory. + + Returns: + A reference to the published release. + + Raises: + TimeFFormatError: If ``manifest.json`` is missing, unreadable, or invalid. + TimeNetFileError: If a listed file is missing or its size differs from the manifest. + OSError: If copying or renaming fails, with the backup path if restoring the old directory also fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by _read_release and publish_directory + manifest = self._read_release(from_directory) + reference = DatasetRef(dataset_id=manifest.dataset_id, version=manifest.metadata.dataset_version) + final = release_dir(self._root, reference) + final.parent.mkdir(parents=True, exist_ok=True) + staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}") + try: + for relpath in (*sorted({part.path for part in manifest.files.all_files()}), MANIFEST_FILE): + target = staging / relpath + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(from_directory / relpath, target) + publish_directory(staging, final) + finally: + shutil.rmtree(staging, ignore_errors=True) + return reference def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: """Open a committed version as a local, random-access handle. diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 4a70fa9b8..35a2740a4 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -5,9 +5,7 @@ Artifact bytes never pass through the API. The service hands back presigned URLs. """ -from collections.abc import Callable, Mapping from pathlib import Path -import shutil import tempfile from typing import BinaryIO, cast @@ -15,10 +13,9 @@ import pyarrow.fs as pafs from timenet.config import settings -from timenet.dataset import TimeFDataset from timenet.errors import TimeNetRegistryError -from timenet.manifest import Manifest, PayloadSource -from timenet.registry._paths import release_dir +from timenet.format.constants import MANIFEST_FILE +from timenet.manifest import Manifest, parse_manifest from timenet.registry.lazy_read import LazyReadRegistry from timenet.registry.remote._download import ProgressCallback, download_version_files from timenet.registry.remote._http import RegistryHttpClient @@ -32,11 +29,10 @@ from timenet.registry.version import DatasetVersion from timenet.types import DatasetMetadata from timenet.types.references import DatasetRef, is_latest -from timenet.writer import TimeFWriter, WriteProgressEvent class RemoteRegistry(LazyReadRegistry): - """Serves datasets from the hosted TimeNet registry over HTTP(S).""" + """Read and publish datasets through the hosted TimeNet HTTP service.""" def __init__( self, @@ -92,10 +88,11 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: Raises: TimeNetDatasetNotFoundError: If the dataset id or version is unknown. - """ # ruff: ignore[docstring-extraneous-exception] (raised by the HTTP client on 404, not directly here) + TimeFFormatError: If the served manifest is not valid. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the HTTP client and parse_manifest resolved = self._resolve_latest(dataset_id) if is_latest(version) else version text = self._http.get_text(f"/datasets/{dataset_id}/{resolved}/manifest") - return Manifest.model_validate_json(text) + return parse_manifest(text, f"the manifest of {dataset_id}@{resolved} in {self}") def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """Open one file of a dataset version for binary reading. @@ -159,78 +156,46 @@ def _download_release( # ruff: ignore[too-many-arguments] self._http, manifest, dataset_id, version, Path(dest_dir), force=force, progress_cb=progress_cb ) - def store( - self, - dataset: TimeFDataset, - *, - force: bool = False, - values_backend: str = "parquet", - progress_cb: Callable[[WriteProgressEvent], None] | None = None, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, - ) -> str: - """Compile a dataset locally and publish it to the remote registry. + def put_release(self, from_directory: Path) -> DatasetRef: + """Publish a release directory to the remote registry. - Compiles to a temporary directory, then runs the service publish flow: POST the manifest, PUT - each declared file to its presigned URL, and finalize. An already-committed version is skipped - unless ``force``. + The directory must contain ``manifest.json`` and every file it lists. Args: - dataset: The populated dataset to store. - force: Publish even if the version is already committed. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). - progress_cb: Optional writer progress callback. - pinned_by: The releases that already read files of this release, with their pins. The - writer publishes only if it reproduced each pinned file. + from_directory: The release directory. Returns: - The stored version string. + A reference to the published release. Raises: - TimeNetRegistryError: If the service requests a file that was not produced locally. - """ - if dataset.schema is None: - dataset.derive_schema() - dataset_id = dataset.metadata.dataset_id - version = str(dataset.metadata.dataset_version) - if not force and self.exists(dataset_id, version): - return version - staging_root = Path(tempfile.mkdtemp(prefix="timenet-publish-")) - try: - # The writer stages locally but checks every payload source against this registry. - with TimeFWriter( - staging_root, - dataset, - values_backend=values_backend, - progress_cb=progress_cb, - registry=self, - pinned_by=pinned_by, - ) as writer: - writer.write() - version_dir = release_dir(staging_root, DatasetRef.parse(f"{dataset_id}@{version}")) - manifest_bytes = (version_dir / "manifest.json").read_bytes() - manifest = Manifest.model_validate_json(manifest_bytes.decode()) - declared = {part.path for part in manifest.files.all_files()} - publish_endpoint = f"/datasets/{dataset_id}/{version}/publish" - publish = PublishResponseModel.model_validate_json( - self._http.post(publish_endpoint, content=manifest_bytes).content + TimeFFormatError: If ``manifest.json`` is missing, unreadable, or invalid. + TimeNetFileError: If a listed file is missing or its size differs from the manifest. + TimeNetRegistryError: If the service skips a listed file or requests a missing file. + """ # ruff: ignore[docstring-extraneous-exception] - raised by _read_release + manifest = self._read_release(from_directory) + dataset_id = manifest.dataset_id + version = str(manifest.metadata.dataset_version) + manifest_bytes = (from_directory / MANIFEST_FILE).read_bytes() + declared = {part.path for part in manifest.files.all_files()} + publish_endpoint = f"/datasets/{dataset_id}/{version}/publish" + publish = PublishResponseModel.model_validate_json( + self._http.post(publish_endpoint, content=manifest_bytes).content + ) + files = publish.files + # Cross-check the service's upload list against the manifest before uploading. A manifest file + # the service omits would publish an incomplete version; a requested file we did not produce is + # not ours to upload. Both fail loudly here instead of partway through the upload. + unrequested = sorted(declared - set(files)) + missing = [relpath for relpath in files if not (from_directory / relpath).is_file()] + if unrequested or missing: + raise TimeNetRegistryError( + f"publish for {dataset_id}@{version} disagrees with the manifest: " + f"service did not request {unrequested}, requested files not produced locally {missing}" ) - files = publish.files - # Cross-check the service's upload list against the manifest before uploading. A manifest file - # the service omits would publish an incomplete version; a requested file we did not produce is - # not ours to upload. Both fail loudly here instead of partway through the upload. - unrequested = sorted(declared - set(files)) - missing = [relpath for relpath in files if not (version_dir / relpath).is_file()] - if unrequested or missing: - raise TimeNetRegistryError( - f"publish for {dataset_id}@{version} disagrees with the manifest: " - f"service did not request {unrequested}, requested files not produced locally {missing}" - ) - for relpath in files: - self._upload_file(dataset_id, version, version_dir, relpath) - self._http.post(f"/datasets/{dataset_id}/{version}/finalize") - finally: - shutil.rmtree(staging_root, ignore_errors=True) - return version + for relpath in files: + self._upload_file(dataset_id, version, from_directory, relpath) + self._http.post(f"/datasets/{dataset_id}/{version}/finalize") + return DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version) def _upload_file(self, dataset_id: str, version: str, version_dir: Path, relpath: str) -> None: """Fetch a presigned PUT URL for one file and upload its bytes. diff --git a/packages/timenet/src/timenet/registry/s3.py b/packages/timenet/src/timenet/registry/s3.py index f35a84ea7..4a7d894e5 100644 --- a/packages/timenet/src/timenet/registry/s3.py +++ b/packages/timenet/src/timenet/registry/s3.py @@ -6,16 +6,13 @@ 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 ``store`` and to fetch by an explicit ``org/name@version``. +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. """ -from collections.abc import Callable, Mapping import os from pathlib import Path -import shutil -import tempfile from typing import Any, BinaryIO from urllib.parse import urlparse import uuid @@ -23,17 +20,14 @@ import pyarrow.fs as pafs from timenet.config import settings -from timenet.dataset import TimeFDataset from timenet.errors import TimeNetDatasetNotFoundError, TimeNetRegistryError from timenet.format.constants import MANIFEST_FILE -from timenet.manifest import Manifest, PayloadSource -from timenet.registry._paths import release_dir +from timenet.manifest import Manifest, parse_manifest from timenet.registry.lazy_read import LazyReadRegistry from timenet.registry.version import DatasetVersion from timenet.types import DatasetMetadata, Version from timenet.types.metadata_fields import validate_dataset_id from timenet.types.references import DatasetRef, is_latest -from timenet.writer import TimeFWriter, WriteProgressEvent # Dataset objects live under a ``datasets/`` prefix, matching the hosted registry's bucket layout, so @@ -44,6 +38,8 @@ class S3Registry(LazyReadRegistry): """Serves and publishes datasets under an ``s3://bucket/prefix`` root.""" + _staging_prefix = "timenet-s3-" + def __init__(self, uri: str, *, cache_dir: str | Path | None = None) -> None: """Open an S3-backed registry. @@ -103,7 +99,8 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: Raises: TimeNetDatasetNotFoundError: If the dataset id or version has no manifest in the bucket. - """ + TimeFFormatError: If the stored manifest is not valid. + """ # ruff: ignore[docstring-extraneous-exception] - raised by parse_manifest validate_dataset_id(dataset_id) resolved = self._latest_version(dataset_id) if is_latest(version) else version if resolved is None: @@ -115,7 +112,7 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: raise TimeNetDatasetNotFoundError( f"no manifest for {dataset_id!r} version {resolved!r} under {self._display_root()}" ) - return Manifest.model_validate_json(body.decode()) + return parse_manifest(body, f"the manifest of {dataset_id}@{resolved} in {self}") def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """Open one file of a dataset version for binary reading. @@ -150,68 +147,37 @@ def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: root = f"{self._bucket}/{self._key(reference.dataset_id, str(reference.version))}" return DatasetVersion(manifest, self._s3_filesystem(), root) - def store( - self, - dataset: TimeFDataset, - *, - force: bool = False, - values_backend: str = "parquet", - progress_cb: Callable[[WriteProgressEvent], None] | None = None, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, - ) -> str: - """Compile a dataset locally and upload it under this registry's prefix. - - Uploads every file to a temporary ``.tmp-`` prefix first. Then moves each object - into the final ``/`` prefix with a server-side copy, writing ``manifest.json`` last as - the commit marker. A failed publish then never leaves a partial version a reader can trust. An - already-committed version is skipped unless ``force``. + def put_release(self, from_directory: Path) -> DatasetRef: + """Upload a release directory to S3. + + The directory must contain ``manifest.json`` and every file it lists. This method uploads + the listed files and publishes ``manifest.json`` last. Args: - dataset: The populated dataset to store. - force: Publish even if the version is already committed. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). - progress_cb: Optional writer progress callback. - pinned_by: The releases that already read files of this release, with their pins. The - writer publishes only if it reproduced each pinned file. + from_directory: The release directory. Returns: - The stored version string. - """ - if dataset.schema is None: - dataset.derive_schema() - dataset_id = dataset.metadata.dataset_id - version = str(dataset.metadata.dataset_version) - if not force and self.exists(dataset_id, version): - return version + A reference to the published release. + + Raises: + TimeFFormatError: If ``manifest.json`` is missing, unreadable, or invalid. + TimeNetFileError: If a listed file is missing or its size differs from the manifest. + """ # ruff: ignore[docstring-extraneous-exception] - raised by _read_release + manifest = self._read_release(from_directory) client = self._client() - final_base = self._key(dataset_id, version) + final_base = self._key(manifest.dataset_id, str(manifest.metadata.dataset_version)) temp_base = f"{final_base}.tmp-{uuid.uuid4().hex}" - staging = Path(tempfile.mkdtemp(prefix="timenet-s3-")) try: - # The writer stages locally but checks every payload source against this registry. - with TimeFWriter( - staging, - dataset, - values_backend=values_backend, - progress_cb=progress_cb, - registry=self, - pinned_by=pinned_by, - ) as writer: - writer.write() - version_dir = release_dir(staging, DatasetRef.parse(f"{dataset_id}@{version}")) self._delete_prefix(client, f"{final_base}.tmp-") # sweep a prior crashed publish's temp prefix - files = sorted(path for path in version_dir.rglob("*") if path.is_file()) - for path in files: - relpath = path.relative_to(version_dir).as_posix() - client.upload_file(str(path), self._bucket, f"{temp_base}/{relpath}") - data_relpaths = [path.relative_to(version_dir).as_posix() for path in files if path.name != MANIFEST_FILE] + data_relpaths = sorted({part.path for part in manifest.files.all_files()}) + for relpath in (*data_relpaths, MANIFEST_FILE): + client.upload_file(str(from_directory / relpath), self._bucket, f"{temp_base}/{relpath}") for relpath in data_relpaths: self._move(client, f"{temp_base}/{relpath}", f"{final_base}/{relpath}") self._move(client, f"{temp_base}/{MANIFEST_FILE}", f"{final_base}/{MANIFEST_FILE}") finally: - shutil.rmtree(staging, ignore_errors=True) self._delete_prefix(client, f"{temp_base}/") # remove any objects left staged in the temp prefix - return version + return DatasetRef(dataset_id=manifest.dataset_id, version=manifest.metadata.dataset_version) @staticmethod def _s3_filesystem() -> pafs.S3FileSystem: diff --git a/packages/timenet/src/timenet/registry/writable.py b/packages/timenet/src/timenet/registry/writable.py index 1842a611d..953ca548c 100644 --- a/packages/timenet/src/timenet/registry/writable.py +++ b/packages/timenet/src/timenet/registry/writable.py @@ -1,27 +1,33 @@ """Define the writable-registry contract for a backend that build publishes datasets into. -This class extends the read-only :class:`~timenet.registry.BaseRegistry` with one write primitive, -:meth:`WritableRegistry.store`. Backends differ only in where a compiled dataset lands, for example a -local directory, an S3 prefix, or a remote service. So ``store`` is the one abstract method, and -:meth:`exists` is shared. +This class extends the read-only :class:`~timenet.registry.BaseRegistry` with two write primitives, +:meth:`WritableRegistry.put_dataset` and :meth:`WritableRegistry.put_release`. Backends differ only in where a +release lands, for example a local directory, an S3 prefix, or a remote service. So these two are the +abstract methods, and :meth:`exists` is shared. """ from abc import ABC, abstractmethod from collections.abc import Callable, Mapping +from pathlib import Path + +import pyarrow.fs as pafs from timenet.dataset import TimeFDataset from timenet.errors import TimeNetDatasetNotFoundError -from timenet.manifest import PayloadSource +from timenet.format.constants import MANIFEST_FILE +from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry.base import BaseRegistry +from timenet.registry.dependencies import check_file_sizes +from timenet.registry.version import DatasetVersion from timenet.types.references import DatasetRef from timenet.writer import WriteProgressEvent class WritableRegistry(BaseRegistry, ABC): - """A registry that build can publish datasets into, not just read from.""" + """A registry that can also publish datasets and release directories.""" @abstractmethod - def store( + def put_dataset( self, dataset: TimeFDataset, *, @@ -29,24 +35,61 @@ def store( values_backend: str = "parquet", progress_cb: Callable[[WriteProgressEvent], None] | None = None, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, - ) -> str: - """Compile a dataset and publish it to this registry. + ) -> DatasetRef: + """Write a dataset to this registry. + + This method derives a schema if the dataset has none. It returns an existing version's + reference unless ``force`` is set. + + Args: + dataset: The dataset to write. + force: Replace an existing version. + values_backend: Storage format for values (``"parquet"`` or ``"zarr"``). + progress_cb: Optional callback for write progress. + pinned_by: Releases that already read this release's files, with the pins that the new files must match. + + Returns: + A reference to the stored release. + """ + + @abstractmethod + def put_release(self, from_directory: Path) -> DatasetRef: + """Publish an existing release to this registry. - If the dataset has no schema, this method derives one first. If a version is already - committed, this method skips it unless ``force`` is set. + The directory must contain ``manifest.json`` and every file it lists. The method publishes only + those files and leaves the directory unchanged. Args: - dataset: The populated dataset to store. - force: Overwrite an already-committed version instead of skipping it. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). - progress_cb: Optional writer progress callback. - pinned_by: The releases that already read files of this release, with their pins. The - writer publishes only if it reproduced each pinned file. + from_directory: The release directory. Returns: - The stored version string. + A reference to the published release. + + Raises: + TimeFFormatError: If ``manifest.json`` is missing, unreadable, or invalid. + TimeNetFileError: If a listed file is missing or its size differs from the manifest. """ + @staticmethod + def _read_release(directory: Path) -> Manifest: + """Read a release manifest and check the listed file sizes. + + Args: + directory: The release directory. + + Returns: + The release manifest. + + Raises: + TimeFFormatError: If ``manifest.json`` is missing, unreadable, or invalid. + TimeNetFileError: If a listed file is missing or its size differs from the manifest. + """ # ruff: ignore[docstring-extraneous-exception] - raised by read_manifest and check_file_sizes + manifest = read_manifest(directory / MANIFEST_FILE) + reference = DatasetRef(dataset_id=manifest.dataset_id, version=manifest.metadata.dataset_version) + version = DatasetVersion(manifest, pafs.LocalFileSystem(), str(directory)) + check_file_sizes(version, manifest.files.all_files(), owner=str(reference)) + return manifest + def exists(self, dataset_id: str, version: str) -> bool: """Return whether a committed version already exists in this registry. diff --git a/packages/timenet/src/timenet/writer/writer.py b/packages/timenet/src/timenet/writer/writer.py index 1a1f0377c..f8e094d92 100644 --- a/packages/timenet/src/timenet/writer/writer.py +++ b/packages/timenet/src/timenet/writer/writer.py @@ -10,6 +10,7 @@ import numpy as np import pyarrow as pa +from timenet._fs import publish_directory from timenet.dataset import Signal, TimeFDataset from timenet.dataset.time_series import _validate_enum_values from timenet.errors import TimeFFormatError, TimeFValidationError @@ -248,32 +249,12 @@ def close(self) -> None: write as pinned. Nothing is published. OSError: If publishing fails and the old release cannot be moved back either. The error names the ``.old-`` directory that holds it. - """ # ruff: ignore[docstring-extraneous-exception] - _check_pinned_by raises TimeFFormatError + """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_pinned_by and publish_directory if not self._written: raise RuntimeError("write() must run successfully before close()") self._check_pinned_by(self._write_manifest()) self._final_dir.parent.mkdir(parents=True, exist_ok=True) - old = None - if self._final_dir.exists(): - # A committed version reaches here only with replace=True, which LocalRegistry.store() - # and run_pipeline() pass for force. Otherwise the caller pre-created the directory. - old = self._final_dir.with_name(f"{self._final_dir.name}.old-{uuid.uuid4().hex}") - self._final_dir.replace(old) - try: - self._staging_dir.replace(self._final_dir) - except BaseException as error: - if old is not None: - try: - old.replace(self._final_dir) - except OSError as restore_error: - raise OSError( - f"could not publish {self._final_dir} ({error}) or move the old release back " - f"({restore_error}); it is in {old}, so rename that to {self._final_dir} to recover it" - ) from error - raise - if old is not None: - # The new release is published, so failing to delete the old one must not fail the write. - shutil.rmtree(old, ignore_errors=True) + publish_directory(self._staging_dir, self._final_dir) self._emit(ProgressStage.COMMIT, 1, 1) def abort(self) -> None: diff --git a/packages/timenet/tests/reader/test_get.py b/packages/timenet/tests/reader/test_get.py index 6efb030f1..72ccb7b18 100644 --- a/packages/timenet/tests/reader/test_get.py +++ b/packages/timenet/tests/reader/test_get.py @@ -448,7 +448,7 @@ def test_a_manifest_with_two_parent_cards_for_one_release_is_refused(child): Manifest.model_validate(doubled) # A release stored with such a manifest fails to open, before a reader reads anything. path.write_text(json.dumps(doubled)) - with pytest.raises(ValidationError, match=match): + with pytest.raises(TimeFFormatError, match=f"is not a valid manifest: {match}"): open_local_version(child) diff --git a/packages/timenet/tests/reader/test_reader.py b/packages/timenet/tests/reader/test_reader.py index 45bc953f0..473138274 100644 --- a/packages/timenet/tests/reader/test_reader.py +++ b/packages/timenet/tests/reader/test_reader.py @@ -13,7 +13,6 @@ import duckdb import pyarrow.fs as pafs import pyarrow.parquet as pq -from pydantic import ValidationError import pytest from timenet.client import TimeNet @@ -545,7 +544,7 @@ def test_unsupported_format_version_raises(tmp_path, format_version): manifest = json.loads(manifest_path.read_text()) manifest["timef_format_version"] = format_version manifest_path.write_text(json.dumps(manifest)) - with pytest.raises(ValidationError, match="timef_format_version"): + with pytest.raises(TimeFFormatError, match="is not a valid manifest: timef_format_version"): open_local_version(version_dir) diff --git a/packages/timenet/tests/reader/test_shared_storage.py b/packages/timenet/tests/reader/test_shared_storage.py index 66e6577e3..55730194a 100644 --- a/packages/timenet/tests/reader/test_shared_storage.py +++ b/packages/timenet/tests/reader/test_shared_storage.py @@ -235,29 +235,29 @@ def test_a_registry_refuses_a_release_whose_payload_sources_it_does_not_hold(tmp match=rf"payload source timenet/hello-world@1\.0\.0 of test/child@1\.0\.0 is not in local registry " rf"{tmp_path / 'children'}; store test/child@1\.0\.0 in the registry that holds", ): - elsewhere.store(child) + elsewhere.put_dataset(child) assert not elsewhere.exists("test/child", "1.0.0") assert not list((tmp_path / "children").rglob("*.tmp-*")) # Next to the release that stores its payloads, the same dataset is stored and opens. - LocalRegistry(tmp_path / "parents").store(child) + LocalRegistry(tmp_path / "parents").put_dataset(child) expected = parent.records[0].signals[0].to_arrow() with TimeFReader(open_local_version(tmp_path / "parents/test/child/1.0.0")) as reader: assert reader.read().records[0].signals[0].to_arrow().equals(expected) -def test_a_refused_forced_store_keeps_the_committed_release(tmp_path): +def test_a_refused_forced_put_dataset_keeps_the_committed_release(tmp_path): parent_dir = store(tmp_path / "parents", make_dataset()) children = LocalRegistry(tmp_path / "children") with TimeFReader(open_local_version(parent_dir)) as reader: parent = reader.read() # The committed release owns its values, so this registry holds everything it needs. - children.store(_release("test/child", [_record_in_memory("old", "Old report.", [0, 1_000])])) + children.put_dataset(_release("test/child", [_record_in_memory("old", "Old report.", [0, 1_000])])) child_dir = tmp_path / "children/test/child/1.0.0" before = (child_dir / "manifest.json").read_bytes() child = _derive(parent, "test/child", [parent.records[0]]) with pytest.raises(TimeNetDatasetNotFoundError, match=r"payload source .* of test/child@1\.0\.0 is not in"): - children.store(child, force=True) + children.put_dataset(child, force=True) assert (child_dir / "manifest.json").read_bytes() == before with TimeFReader(open_local_version(child_dir)) as reader: assert [_name(record) for record in reader.read().records] == ["old"] @@ -387,7 +387,7 @@ def test_a_child_pins_exactly_the_source_files_it_reads(tmp_path, backend): def test_opening_a_child_detects_a_forced_replace_of_its_source(tmp_path): _, child_dir = _parent_and_child(tmp_path) - LocalRegistry(tmp_path).store( + LocalRegistry(tmp_path).put_dataset( _release("test/p", [_record_in_memory("p", "Replaced. " * 300, [0, 2_000])]), force=True ) first = min(part.path for part in _pins(child_dir)["test/p@1.0.0"].files) @@ -403,7 +403,7 @@ def test_storing_a_child_detects_a_forced_replace_of_its_source(tmp_path): registry = LocalRegistry(tmp_path) with TimeFReader(open_local_version(parent_dir)) as reader: loaded = reader.read() - registry.store(_release("test/p", [_record_in_memory("p", "Replaced. " * 300, [0, 2_000])]), force=True) + registry.put_dataset(_release("test/p", [_record_in_memory("p", "Replaced. " * 300, [0, 2_000])]), force=True) # Both releases store the same values, so their shards match; the time offsets differ, and the # error names that payload file. with pytest.raises( @@ -412,23 +412,23 @@ def test_storing_a_child_detects_a_forced_replace_of_its_source(tmp_path): r"values were read from: they read 'payloads/part-00000000\.parquet' as \d+ bytes with checksum " r"sha256:[0-9a-f]{64}, but test/p@1\.0\.0 there records \d+ bytes with checksum sha256:[0-9a-f]{64}$", ): - registry.store(_derive(loaded, "test/c", loaded.records)) + registry.put_dataset(_derive(loaded, "test/c", loaded.records)) assert not (tmp_path / "test/c").exists() or not list((tmp_path / "test/c").iterdir()) def test_a_child_is_refused_by_a_registry_that_holds_another_release_of_its_source(tmp_path): ours, foreign = LocalRegistry(tmp_path / "ours"), LocalRegistry(tmp_path / "foreign") - ours.store(_release("test/p", [_record_in_memory("p", "Ours. " * 300, [0, 1_000])])) - foreign.store(_release("test/p", [_record_in_memory("p", "Foreign. " * 300, [0, 3_000])])) + 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: loaded = reader.read() with pytest.raises( TimeFFormatError, match=r"payload source test/p@1\.0\.0 in local registry .*foreign differs from the release", ): - foreign.store(_derive(loaded, "test/c", loaded.records)) + foreign.put_dataset(_derive(loaded, "test/c", loaded.records)) assert not foreign.exists("test/c", "1.0.0") - ours.store(_derive(loaded, "test/c", loaded.records)) + ours.put_dataset(_derive(loaded, "test/c", loaded.records)) assert _reports(tmp_path / "ours/test/c/1.0.0") == {"p": ([0.0, 1.0], [0, 1_000], "Ours. " * 300)} @@ -475,7 +475,7 @@ def test_a_release_cannot_reuse_payloads_of_a_source_that_reads_it(tmp_path): with pytest.raises( TimeFValidationError, match=r"test/c@1\.0\.0, which reads payloads of test/p@1\.0\.0 itself" ): - registry.store(_derive(loaded, "test/p", own), force=True) + registry.put_dataset(_derive(loaded, "test/p", own), force=True) assert {name: _reports(directory) for name, directory in (("p", parent_dir), ("c", child_dir))} == before diff --git a/packages/timenet/tests/registry/test_conformance.py b/packages/timenet/tests/registry/test_conformance.py index a4230aa39..1fe6a7873 100644 --- a/packages/timenet/tests/registry/test_conformance.py +++ b/packages/timenet/tests/registry/test_conformance.py @@ -1,7 +1,7 @@ """One contract, three backends. Every writable registry must behave the same way from a caller's seat. The ``registry`` fixture parametrizes over ``local``, ``s3``, and ``remote``. Each yields the same two -datasets, published by calling ``store``, so a single test body runs against all three. Transport-specific +datasets, published by calling ``put_dataset``, so a single test body runs against all three. Transport-specific behaviour (S3 key prefixes, remote 429/auth/presign, factory dispatch) stays in the per-backend files; this file only asserts the shared read/write contract. """ @@ -14,7 +14,7 @@ from timenet.dataset import Record, Signal, Source, TimeFDataset from timenet.dataset.axis import RegularAxis -from timenet.errors import TimeNetDatasetNotFoundError +from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError from timenet.registry import LocalRegistry, RemoteRegistry, S3Registry, WritableRegistry from timenet.testing import make_dataset, sine_loader from timenet.types import ( @@ -29,6 +29,7 @@ Version, ureg, ) +from timenet.types.references import DatasetRef RHYTHM = LabelSet( @@ -102,7 +103,7 @@ def _s3_registry(request, tmp_path, monkeypatch) -> S3Registry: @pytest.fixture(params=["local", "s3", "remote"]) def backend(request, tmp_path, monkeypatch) -> Backend: - """A writable registry of the requested kind, seeded via ``store``. + """A writable registry of the requested kind, seeded via ``put_dataset``. Holds timenet/hello-world at 1.0.0 and 1.1.0 (so latest-resolution has a real choice) plus a second dataset demo/other, so listing has more than one id to sort. @@ -123,7 +124,7 @@ def backend(request, tmp_path, monkeypatch) -> Backend: newer = make_dataset() newer._metadata = newer.metadata.model_copy(update={"dataset_version": Version(1, 1, 0)}) for dataset in (make_dataset(), newer, _second_dataset()): - registry.store(dataset) + registry.put_dataset(dataset) return Backend(registry=registry, supports_listing=supports_listing) @@ -190,16 +191,51 @@ def test_exists_false_for_unknown_version(backend): assert not backend.registry.exists("timenet/hello-world", "9.9.9") -# ---- store ------------------------------------------------------------------------------------ +# ---- put_dataset ------------------------------------------------------------------------------ -def test_store_is_idempotent(backend): - # The version is already committed by the fixture; a second store must skip, not raise. - assert backend.registry.store(make_dataset()) == "1.0.0" +def test_put_dataset_is_idempotent(backend): + # The version is already committed by the fixture; a second put_dataset() call must skip, not raise. + assert backend.registry.put_dataset(make_dataset()) == DatasetRef.parse("timenet/hello-world@1.0.0") -def test_store_force_republishes(backend): - assert backend.registry.store(make_dataset(), force=True) == "1.0.0" +def test_put_dataset_force_republishes(backend): + assert backend.registry.put_dataset(make_dataset(), force=True) == DatasetRef.parse("timenet/hello-world@1.0.0") + + +def _local_release(tmp_path): + """Build a complete local release of timenet/hello-world 2.0.0; return its directory and manifest.""" + source = LocalRegistry(tmp_path / "src") + dataset = make_dataset() + dataset._metadata = dataset.metadata.model_copy(update={"dataset_version": Version(2, 0, 0)}) + source.put_dataset(dataset) + return source.root / "timenet/hello-world/2.0.0", source.get_manifest("timenet/hello-world", "2.0.0") + + +def test_put_release_publishes_a_complete_release(backend, tmp_path): + directory, manifest = _local_release(tmp_path) + assert backend.registry.put_release(directory) == DatasetRef.parse("timenet/hello-world@2.0.0") + assert backend.registry.get_manifest("timenet/hello-world", "2.0.0") == manifest + + +def test_put_release_publishes_only_the_declared_files(backend, tmp_path): + directory, manifest = _local_release(tmp_path) + (directory / "stray.txt").write_text("not part of the release") + (directory / "time_series" / "stray.parquet").write_text("not part of the release") + backend.registry.put_release(directory) + assert backend.registry.get_manifest("timenet/hello-world", "2.0.0") == manifest + for relpath in ("stray.txt", "time_series/stray.parquet"): + with pytest.raises((FileNotFoundError, TimeNetDatasetNotFoundError)): + backend.registry.open_file("timenet/hello-world", "2.0.0", relpath).close() + + +def test_put_release_refuses_an_incomplete_directory(backend, tmp_path): + directory, manifest = _local_release(tmp_path) + declared = sorted(part.path for part in manifest.files.all_files()) + (directory / declared[0]).unlink() + with pytest.raises(TimeNetFileError): + backend.registry.put_release(directory) + assert not backend.registry.exists("timenet/hello-world", "2.0.0") # ---- list_datasets (catalog-only) ------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py index 5bdbc8d42..00cf716f8 100644 --- a/packages/timenet/tests/registry/test_registry.py +++ b/packages/timenet/tests/registry/test_registry.py @@ -1,9 +1,11 @@ from pathlib import Path +import re from pydantic import ValidationError import pytest from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetRegistryError +from timenet.manifest import Manifest from timenet.reader import TimeFReader from timenet.registry import ( TIMENET_REGISTRY_URL, @@ -20,6 +22,7 @@ ) from timenet.testing import make_dataset from timenet.types import Annotation, AnswerTask, Domain, License +from timenet.types.references import DatasetRef from timenet.writer import TimeFWriter @@ -142,6 +145,13 @@ def test_get_manifest_rejects_mismatched_id(registry_root): registry.get_manifest("wrong/place") +def test_get_manifest_names_the_path_of_a_cut_manifest(registry_root): + path = registry_root / "timenet/hello-world/1.0.0/manifest.json" + path.write_text(path.read_text()[:100]) + with pytest.raises(TimeFFormatError, match=f"{re.escape(str(path))} is not a valid manifest: Invalid JSON"): + LocalRegistry(registry_root).get_manifest("timenet/hello-world", "1.0.0") + + def test_latest_version_ignores_staging_dirs(registry_root): # A crashed build can leave a `.tmp-` sibling (briefly holding a manifest.json). stale = registry_root / "demo/ecg" / "2.0.0.tmp-deadbeef" @@ -249,22 +259,22 @@ def test_search_negative_limit_rejected(registry_root): LocalRegistry(registry_root).search(limit=-1) -# ---- store (write side) ----------------------------------------------------------------------- +# ---- put_dataset (write side) ----------------------------------------------------------------- -def test_store_derives_schema_if_absent(tmp_path): +def test_put_dataset_derives_schema_if_absent(tmp_path): dataset = make_dataset() assert dataset.schema is None - LocalRegistry(tmp_path).store(dataset) + LocalRegistry(tmp_path).put_dataset(dataset) assert dataset.schema is not None -def test_a_forced_store_replaces_the_committed_release(tmp_path): +def test_a_forced_put_dataset_replaces_the_committed_release(tmp_path): registry = LocalRegistry(tmp_path) - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) replacement = make_dataset() replacement.records[0].annotate(Annotation(key="replaced", value=1)) - registry.store(replacement, force=True) + registry.put_dataset(replacement, force=True) version_dir = tmp_path / "timenet/hello-world/1.0.0" with TimeFReader(open_local_version(version_dir)) as reader: @@ -273,9 +283,9 @@ def test_a_forced_store_replaces_the_committed_release(tmp_path): assert [path.name for path in version_dir.parent.iterdir()] == ["1.0.0"] -def test_a_failed_forced_store_keeps_the_committed_release(tmp_path, monkeypatch): +def test_a_failed_forced_put_dataset_keeps_the_committed_release(tmp_path, monkeypatch): registry = LocalRegistry(tmp_path) - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) version_dir = tmp_path / "timenet/hello-world/1.0.0" before = (version_dir / "manifest.json").read_bytes() @@ -284,11 +294,46 @@ def fail(self): monkeypatch.setattr(TimeFWriter, "write", fail) with pytest.raises(RuntimeError, match="write failed"): - registry.store(make_dataset(), force=True) + registry.put_dataset(make_dataset(), force=True) assert (version_dir / "manifest.json").read_bytes() == before assert [path.name for path in version_dir.parent.iterdir()] == ["1.0.0"] +def _release_directory(root: Path, *, annotation: str | None = None) -> tuple[Path, Manifest]: + """Store timenet/hello-world 1.0.0 under ``root``; return its directory and manifest.""" + dataset = make_dataset() + if annotation is not None: + dataset.records[0].annotate(Annotation(key=annotation, value=1)) + source = LocalRegistry(root) + source.put_dataset(dataset) + return root / "timenet/hello-world/1.0.0", source.get_manifest("timenet/hello-world", "1.0.0") + + +def test_put_release_replaces_a_committed_copy(tmp_path): + registry = LocalRegistry(tmp_path / "reg") + first, _ = _release_directory(tmp_path / "first") + registry.put_release(first) + directory, manifest = _release_directory(tmp_path / "second", annotation="replaced") + assert registry.put_release(directory) == DatasetRef.parse("timenet/hello-world@1.0.0") + # A fresh registry, so that no cached manifest of the first copy answers. + assert LocalRegistry(registry.root).get_manifest("timenet/hello-world", "1.0.0") == manifest + version_dir = registry.root / "timenet/hello-world/1.0.0" + assert [path.name for path in version_dir.parent.iterdir()] == ["1.0.0"] # no .tmp-* or .old-* left + + +@pytest.mark.parametrize("manifest_text", [None, "{not json"]) +def test_put_release_refuses_a_directory_without_a_readable_manifest(tmp_path, manifest_text): + directory, _ = _release_directory(tmp_path / "source") + if manifest_text is None: + (directory / "manifest.json").unlink() + else: + (directory / "manifest.json").write_text(manifest_text) + registry = LocalRegistry(tmp_path / "reg") + with pytest.raises(TimeFFormatError, match=re.escape(str(directory / "manifest.json"))): + registry.put_release(directory) + assert not registry.root.exists() + + # ---- s3 catalog ------------------------------------------------------------------------------- diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index 5552c2cdb..69e96c072 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -375,20 +375,19 @@ def serve_file(request): assert len(server.responses) == transferred -def test_store_publishes_uploads_and_finalizes(tmp_path): +def test_put_dataset_publishes_uploads_and_finalizes(tmp_path): transport, state = build_publish_fake(token="tok_rw") registry = RemoteRegistry("http://api.local", token="tok_rw", transport=transport, cache_dir=tmp_path) dataset = make_dataset() - version = registry.store(dataset) - assert version == str(dataset.metadata.dataset_version) + assert registry.put_dataset(dataset) == DatasetRef.parse(f"{dataset.metadata.dataset_id}@1.0.0") assert state["finalized"] is True # every declared file was uploaded assert set(state["store"]) == set(state["published"]) -def test_store_refuses_a_payload_source_the_service_does_not_hold_before_publishing(tmp_path): +def test_put_dataset_refuses_a_payload_source_the_service_does_not_hold_before_publishing(tmp_path): local = LocalRegistry(tmp_path / "local") - local.store(make_dataset()) + 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: @@ -403,16 +402,16 @@ def test_store_refuses_a_payload_source_the_service_does_not_hold_before_publish with pytest.raises( TimeNetDatasetNotFoundError, match=r"of test/child@1\.0\.0 is not in registry http://api\.local" ): - registry.store(child) + registry.put_dataset(child) assert state["manifest"] is None assert not state["store"] -def test_store_rejects_a_publish_list_that_drops_a_manifest_file(tmp_path): +def test_put_dataset_rejects_a_publish_list_that_drops_a_manifest_file(tmp_path): import httpx # ruff: ignore[import-outside-top-level] def handler(request): - # The service omits every declared file from its upload list. store() must reject that before + # The service omits every declared file from its upload list. put_dataset() must reject that before # uploading, rather than publish an incomplete version. if request.url.path.endswith("/publish"): return httpx.Response(200, json={"files": []}) @@ -422,10 +421,10 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(TimeNetRegistryError, match="disagrees with the manifest"): - registry.store(make_dataset(), force=True) + registry.put_dataset(make_dataset(), force=True) -def test_store_rejects_unsafe_publish_paths(tmp_path): +def test_put_dataset_rejects_unsafe_publish_paths(tmp_path): def handler(request): if request.url.path.endswith("/publish"): return httpx.Response(200, json={"files": ["../escape"]}) @@ -435,10 +434,10 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(ValidationError, match="stay within the dataset root"): - registry.store(make_dataset(), force=True) + registry.put_dataset(make_dataset(), force=True) -def test_store_rejects_malformed_upload_grant(tmp_path): +def test_put_dataset_rejects_malformed_upload_grant(tmp_path): def handler(request): if request.url.path.endswith("/publish"): manifest = json.loads(request.content) @@ -452,7 +451,7 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(ValidationError, match="headers"): - registry.store(make_dataset(), force=True) + registry.put_dataset(make_dataset(), force=True) def test_latest_detail_rejects_non_string_version(tmp_path): @@ -473,11 +472,11 @@ def test_remote_registry_uses_client_storage_path(tmp_path): assert client._registry._cache_dir == client._storage == tmp_path / "store" -def test_store_without_write_token_is_rejected(tmp_path): +def test_put_dataset_without_write_token_is_rejected(tmp_path): transport, _ = build_publish_fake(token="tok_rw") registry = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path) # anonymous with pytest.raises(TimeNetRegistryError): - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) @pytest.mark.parametrize("backend", ["parquet", "zarr"]) @@ -487,7 +486,7 @@ def test_child_streams_physical_sources_and_downloads_for_offline_reads(tmp_path parent = make_dataset() build_only = parent.metadata.model_copy(update={"dataset_id": "test/build-only"}) parent._import_state.parent_cards[ParentAlias("build-only")] = ParentDatasetCard(metadata=build_only) - local.store(parent, values_backend=backend) + 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: reference = DatasetRef(dataset_id=parent_id, version=parent.metadata.dataset_version) @@ -499,7 +498,7 @@ def test_child_streams_physical_sources_and_downloads_for_offline_reads(tmp_path imported = child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) expected = parent.records[0].signals[0].to_arrow() signal_id = imported.records[parent.records[0].id].signals[0].id - local.store(child) + local.put_dataset(child) requests = [] def handler(request): diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py index 81b26b690..e7857904b 100644 --- a/packages/timenet/tests/registry/test_s3.py +++ b/packages/timenet/tests/registry/test_s3.py @@ -17,6 +17,7 @@ from timenet.registry import LocalRegistry, S3Registry # 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 # ruff: ignore[module-import-not-at-top-of-file] +from timenet.types.references import DatasetRef # ruff: ignore[module-import-not-at-top-of-file] @pytest.fixture @@ -45,10 +46,10 @@ def s3_root(monkeypatch): @pytest.mark.parametrize("entry_point", ["open", "load"]) @pytest.mark.parametrize("backend", ["parquet", "zarr"]) -def test_store_and_remote_load_round_trip(s3_root, tmp_path, backend, entry_point): +def test_put_dataset_and_remote_load_round_trip(s3_root, tmp_path, backend, entry_point): registry = S3Registry(s3_root, cache_dir=tmp_path / "cache") - version = registry.store(make_dataset(), values_backend=backend) - assert version == str(make_dataset().metadata.dataset_version) + reference = registry.put_dataset(make_dataset(), values_backend=backend) + assert reference == DatasetRef.parse(f"{make_dataset().metadata.dataset_id}@1.0.0") client = TimeNet(registry=registry, storage_path=tmp_path / "cache") if entry_point == "open": with client.open(make_dataset().metadata.dataset_id) as reader: @@ -61,9 +62,9 @@ def test_store_and_remote_load_round_trip(s3_root, tmp_path, backend, entry_poin assert not (tmp_path / "cache" / make_dataset().metadata.dataset_id).exists() -def test_store_checks_payload_sources_against_the_bucket(s3_root, tmp_path): +def test_put_dataset_checks_payload_sources_against_the_bucket(s3_root, tmp_path): local = LocalRegistry(tmp_path / "local") - local.store(make_dataset()) + 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: @@ -79,7 +80,7 @@ def test_store_checks_payload_sources_against_the_bucket(s3_root, tmp_path): TimeNetDatasetNotFoundError, match=r"of test/child@1\.0\.0 is not in S3 registry s3://tn-test/payload-sources", ): - registry.store(child) + registry.put_dataset(child) assert not registry.exists("test/child", "1.0.0") keys = [obj["Key"] for obj in boto3.client("s3").list_objects_v2(Bucket="tn-test").get("Contents", [])] assert not [key for key in keys if "test/child" in key] @@ -89,7 +90,7 @@ def test_objects_live_under_a_datasets_prefix(s3_root, tmp_path): # Keys sit under /datasets/... so the bucket matches the hosted registry's layout and can # hold other top-level prefixes beside the datasets. registry = S3Registry(s3_root, cache_dir=tmp_path / "cache") - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) keys = [obj["Key"] for obj in boto3.client("s3").list_objects_v2(Bucket="tn-test")["Contents"]] assert keys # something was stored assert all(key.startswith("registry/datasets/") for key in keys) @@ -98,7 +99,7 @@ def test_objects_live_under_a_datasets_prefix(s3_root, tmp_path): def test_download_then_load_serves_from_cache(s3_root, tmp_path): registry = S3Registry(s3_root, cache_dir=tmp_path / "cache") dataset_id = make_dataset().metadata.dataset_id - registry.store(make_dataset()) + registry.put_dataset(make_dataset()) 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() diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index e02e77f25..3ea6740b2 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -41,7 +41,7 @@ def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: """ parent = make_dataset() registry = LocalRegistry(root) - registry.store(parent) + registry.put_dataset(parent) reference = DatasetRef(dataset_id=parent.metadata.dataset_id, version=parent.metadata.dataset_version) child = TimeFDataset( metadata=parent.metadata.model_copy( @@ -50,7 +50,7 @@ def _child_of_the_demo(root: Path) -> tuple[LocalRegistry, DatasetRef]: ) with TimeNet(registry).open(parent.metadata.dataset_id) as reader: child.import_from(reader, alias=ParentAlias("parent"), records=[parent.records[0].id]) - registry.store(child) + registry.put_dataset(child) return registry, reference @@ -82,7 +82,7 @@ def build(self, dataset_id, root, *, force=False, build_parents=False, fetch_fro self.calls.append((dataset_id, force, build_parents, fetch_from, pinned_by)) dataset = make_dataset() dataset._metadata = self.card(dataset_id) - LocalRegistry(root).store(dataset, force=force, values_backend=self.BACKEND, pinned_by=pinned_by) + LocalRegistry(root).put_dataset(dataset, force=force, values_backend=self.BACKEND, pinned_by=pinned_by) return root / dataset_id / self.VERSION @@ -151,7 +151,7 @@ def compose(self, raw_refs, context): def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_path): origin = LocalRegistry(tmp_path / "origin") - origin.store(make_dataset()) + origin.put_dataset(make_dataset()) metadata = _Resampled().metadata() reference = metadata.parents[ParentAlias("parent")] run_pipeline(_Resampled(), origin.root, cache_dir=tmp_path / "build-cache") diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index d0ab1dc58..2fb1f96bd 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -112,8 +112,9 @@ def __init__(self) -> None: def exists(self, dataset_id: str, version: str) -> bool: return False - def store(self, dataset, *, force: bool, values_backend: str, progress_cb) -> None: + def put_dataset(self, dataset, *, force: bool, values_backend: str, progress_cb, pinned_by) -> DatasetRef: self.values_backend = values_backend + return DatasetRef.parse(f"{dataset.metadata.dataset_id}@{dataset.metadata.dataset_version}") class _ZarrConnector(_DemoConnector): values_backend = ValuesBackend.ZARR @@ -243,7 +244,7 @@ def _store_credentialed_parent(root: Path) -> None: parent._metadata = parent.metadata.model_copy( update={"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} ) - LocalRegistry(root).store(parent) + LocalRegistry(root).put_dataset(parent) class _ChildConnector(BaseConnector[str]): @@ -350,7 +351,7 @@ def test_a_dataset_with_another_card_than_the_engine_read_fails_and_stores_no_re tmp_path, pipeline, connector, differ ): root = tmp_path / "root" - LocalRegistry(root).store(make_dataset()) + LocalRegistry(root).put_dataset(make_dataset()) before = sorted(root.rglob("*")) with pytest.raises( TimeFValidationError, @@ -371,7 +372,7 @@ def test_run_pipeline_fails_on_a_missing_parent_before_downloading(tmp_path, mon def test_open_parents_closes_the_parents_it_opened_when_a_later_one_is_missing(tmp_path, monkeypatch): - LocalRegistry(tmp_path).store(make_dataset()) + LocalRegistry(tmp_path).put_dataset(make_dataset()) missing = DatasetRef.parse("test/missing@1.0.0") metadata = ( _ChildConnector() @@ -397,7 +398,7 @@ def test_a_parent_reader_that_fails_to_close_does_not_keep_the_other_open(tmp_pa second = make_dataset() second._metadata = second.metadata.model_copy(update={"dataset_id": "test/second"}) for dataset in (make_dataset(), second): - LocalRegistry(tmp_path).store(dataset) + LocalRegistry(tmp_path).put_dataset(dataset) parents = {ParentAlias("parent"): _PARENT, ParentAlias("second"): DatasetRef.parse("test/second@1.0.0")} context = open_parents( _ChildConnector().metadata().model_copy(update={"parents": parents}), LocalRegistry(tmp_path) @@ -426,7 +427,7 @@ def test_a_build_context_made_directly_closes_its_parent_readers(tmp_path, monke second._metadata = second.metadata.model_copy(update={"dataset_id": "test/second"}) registry = LocalRegistry(tmp_path) for dataset in (make_dataset(), second): - registry.store(dataset) + 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")), @@ -492,7 +493,7 @@ def exists(self, dataset_id, version): def open_version(self, dataset_id, version=None): return self.local.open_version(dataset_id, version) - def store(self, dataset, **kwargs): + def put_dataset(self, dataset, **kwargs): pytest.fail("wrote a child whose parent is credentialed") diff --git a/packages/timenet/tests/test_fs.py b/packages/timenet/tests/test_fs.py new file mode 100644 index 000000000..6fcd240b8 --- /dev/null +++ b/packages/timenet/tests/test_fs.py @@ -0,0 +1,49 @@ +from pathlib import Path + +import pytest + +from timenet._fs import publish_directory + + +def _directories(tmp_path: Path) -> tuple[Path, Path]: + """Make an old ``final`` and a new ``staging`` directory, each with one marker file.""" + final = tmp_path / "1.0.0" + final.mkdir() + (final / "marker").write_text("old") + staging = tmp_path / "1.0.0.tmp-new" + staging.mkdir() + (staging / "marker").write_text("new") + return staging, final + + +def test_a_stopped_rename_moves_the_old_copy_back(tmp_path, monkeypatch): + staging, final = _directories(tmp_path) + rename = Path.replace + + def replace(self, target): + if self == staging: + raise KeyboardInterrupt + return rename(self, target) + + monkeypatch.setattr(Path, "replace", replace) + with pytest.raises(KeyboardInterrupt): + publish_directory(staging, final) + assert (final / "marker").read_text() == "old" + assert not list(tmp_path.glob("1.0.0.old-*")) + + +def test_a_failed_restore_names_where_the_old_copy_is(tmp_path, monkeypatch): + staging, final = _directories(tmp_path) + rename = Path.replace + + def replace(self, target): + if self == staging or ".old-" in self.name: + raise OSError("rename failed") + return rename(self, target) + + monkeypatch.setattr(Path, "replace", replace) + with pytest.raises(OSError, match=r"it is in .*/1\.0\.0\.old-[0-9a-f]+, so rename that to") as raised: + publish_directory(staging, final) + (old,) = tmp_path.glob("1.0.0.old-*") + assert str(old) in str(raised.value) + assert (old / "marker").read_text() == "old" diff --git a/packages/timenet/tests/writer/test_writer.py b/packages/timenet/tests/writer/test_writer.py index 2f356c63d..68ca477e2 100644 --- a/packages/timenet/tests/writer/test_writer.py +++ b/packages/timenet/tests/writer/test_writer.py @@ -1,6 +1,5 @@ from dataclasses import dataclass import json -from pathlib import Path import shutil import duckdb @@ -250,44 +249,6 @@ def test_replace_publishes_over_a_committed_version_and_drops_the_old_one(tmp_pa assert [path.name for path in version_dir.parent.iterdir()] == [version_dir.name] -def test_a_failed_publish_rename_moves_the_old_release_back(tmp_path, monkeypatch): - version_dir = _written(tmp_path) - before = (version_dir / "manifest.json").read_bytes() - rename = Path.replace - - def replace(self, target): - if ".tmp-" in self.name: - raise OSError("rename failed") - return rename(self, target) - - monkeypatch.setattr(Path, "replace", replace) - with pytest.raises(OSError, match="rename failed"), TimeFWriter(tmp_path, _replacement(), replace=True) as writer: - writer.write() - assert (version_dir / "manifest.json").read_bytes() == before - assert [path.name for path in version_dir.parent.iterdir()] == [version_dir.name] - - -def test_a_failed_publish_that_cannot_restore_the_old_release_names_where_it_is(tmp_path, monkeypatch): - version_dir = _written(tmp_path) - before = (version_dir / "manifest.json").read_bytes() - rename = Path.replace - - def replace(self, target): - if ".tmp-" in self.name or ".old-" in self.name: - raise OSError("rename failed") - return rename(self, target) - - monkeypatch.setattr(Path, "replace", replace) - with ( - pytest.raises(OSError, match=r"it is in .*/1\.0\.0\.old-[0-9a-f]+, so rename that to") as raised, - TimeFWriter(tmp_path, _replacement(), replace=True) as writer, - ): - writer.write() - (old,) = version_dir.parent.glob(f"{version_dir.name}.old-*") - assert str(old) in str(raised.value) - assert (old / "manifest.json").read_bytes() == before - - def test_the_staging_sweep_keeps_an_old_release_that_a_hard_kill_left(tmp_path): version_dir = _written(tmp_path) old = version_dir.with_name(f"{version_dir.name}.old-deadbeef")