feat(registry)!: copy a release between registries

packages/timenet/src/timenet/_fs.py

@@ -1,5 +1,7 @@
 """Replace a directory with a complete staging directory."""
 
+from collections.abc import Generator
+from contextlib import contextmanager
 from pathlib import Path
 import shutil
 import uuid
@@ -37,3 +39,28 @@ def publish_directory(staging: Path, final: Path) -> None:
     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)
+
+
+@contextmanager
+def staged_directory(final: Path) -> Generator[Path]:
+    """Yield an empty staging directory beside ``final``, and publish it as ``final`` if the block succeeds.
+
+    The old ``final`` stays in place until the new directory is complete. The staging directory is
+    removed in every case.
+
+    Args:
+        final: The destination directory.
+
+    Yields:
+        The staging directory.
+    """
+    final.parent.mkdir(parents=True, exist_ok=True)
+    # A hard kill can leave a staging directory behind. Do not sweep other <name>.tmp-* directories:
+    # a concurrent write of the same release may still use one.
+    staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}")
+    staging.mkdir()
+    try:
+        yield staging
+        publish_directory(staging, final)
+    finally:
+        shutil.rmtree(staging, ignore_errors=True)

packages/timenet/src/timenet/client.py

@@ -308,20 +308,16 @@ def load(
             return reader.read(load=LoadPlan.ALL_EAGER)
 
     def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None:
-        """Raise for a non-open dataset read from a registry that does not host its data.
-
-        Credentialed and restricted datasets cannot be redistributed, so TimeNet never serves their
-        bytes. Such a dataset is build-your-own: a local registry holds only what the user built, so a
-        local read is fine, but a hosted registry can only point the user at where to obtain access.
+        """Check dataset access against the registry's requirements.
 
         Args:
-            dataset_id: The bare dataset id.
-            version: The resolved version, or ``None`` for the latest.
+            dataset_id: The dataset ID without a version suffix.
+            version: The version string, or ``None`` for the latest.
 
         Raises:
-            TimeNetAccessError: If the dataset is not open and the registry is not a local one.
+            TimeNetAccessError: If the dataset or its lineage lacks open access and the registry requires it.
         """
-        if isinstance(self._registry, LocalRegistry):
+        if not self._registry.requires_open_access:
             return
         try:
             manifest = self._registry.get_manifest(dataset_id, version)
@@ -394,28 +390,25 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str |
 def _prepare(
     registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None
 ) -> DatasetVersion:
-    """Prepare a local version and its payload sources in the order that :meth:`TimeNet.open` names.
+    """Prepare a local release and its payload sources.
 
-    If a payload source is still missing after the copy from ``fetch_from``, this function builds it
-    with its own connector. That connector is the backend whose card declares the exact version of
-    the source. The version reads files of the source, so the build gets the pins of the version.
-    The build publishes the source only if it reproduces the pinned files.
+    Try local copies, then ``fetch_from``, then a build if ``auto_build`` is true.
+    A built payload source must match the files that the release pins.
 
     Returns:
-        The version with each payload source attached.
+        The release with each payload source attached.
 
     Raises:
-        TimeNetMissingParentError: If a payload source is absent and this function cannot copy or
-            build it.
-        TimeNetInvalidCardError: If the card of a connector cannot be read.
+        TimeNetMissingParentError: If a payload source is missing and cannot be copied or built.
+        TimeNetInvalidCardError: If a connector's dataset card cannot be read.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by _obtain and exact_builder
     try:
         manifest = registry.get_manifest(dataset_id, version)
     except TimeNetDatasetNotFoundError as miss:
         manifest = _obtain(registry, dataset_id, version, miss, auto_build=auto_build, fetch_from=fetch_from)
     reference = DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version)
     # A registry that is not local serves only releases with access declared ``open``.
-    check = None if fetch_from is None or isinstance(fetch_from, LocalRegistry) else _require_open
+    check = _require_open if fetch_from is not None and fetch_from.requires_open_access else None
     built: set[DatasetRef] = set()
     while True:
         try:

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

@@ -198,52 +198,29 @@ def plan_parents(
     build_parents: bool = False,
     force_build: bool = False,
 ) -> list[ParentStep]:
-    """Plan how to get each parent release that a build of ``dataset_id`` needs.
-
-    For each declared parent, a build needs the parent and the payload sources of that parent. It
-    does not need the payload sources of those payload sources. The plan visits the declared parents
-    of a release only if the plan builds that release. So if the output already holds the requested
-    release, the plan visits its parents only if ``force_build`` is set. If ``force_build`` is not
-    set, the plan visits only the payload sources of that release.
-
-    If the output does not hold a release, the plan fetches it from ``fetch_from``. If ``fetch_from``
-    does not hold it either, the plan builds it with the installed connector that declares that
-    exact version. If a local output holds a damaged copy, the plan fetches the release again and
-    does not build it. A registry that is not local holds only releases whose cards declare
-    ``access: open``. So for such an output, each release that the plan visits must declare
-    ``access: open``. If ``fetch_from`` is not local, each release that the plan fetches from it
-    must declare ``access: open`` too.
-
-    The plan works in two phases. First, a walk from the requested release finds the graph of the
-    releases that the build needs. Then :class:`graphlib.TopologicalSorter` orders the graph and
-    finds cycles, and the plan makes its steps in that order.
+    """Plan how to fetch or build the releases a dataset needs.
+
+    Prefer complete copies, then ``fetch_from``, then builds if ``build_parents`` is true.
+    An existing release needs only its payload sources unless ``force_build`` is set.
 
     Args:
         dataset_id: The dataset to build.
-        output: The registry that the build writes to.
-        fetch_from: The registry to fetch missing parents from. If it is ``None``, the plan fetches
-            nothing.
+        output: The registry to build into.
+        fetch_from: The registry to fetch missing or damaged parents from, or ``None``.
         build_parents: Build missing parents with their installed connectors.
-        force_build: Build the requested release again, even if the output holds it, and plan its
-            parents.
+        force_build: Plan the requested release's parents even if the output already holds the release.
 
     Returns:
-        The steps to fetch or build parents. Each step comes after the steps of the releases that it
-        depends on.
+        Fetch or build steps, ordered with dependencies first.
 
     Raises:
-        LookupError: If no installed connector declares a card for ``dataset_id``.
-        TimeNetInvalidCardError: If the card of an installed connector cannot be read.
-        TimeFValidationError: If the parents form a cycle, if an installed connector declares another
-            version than a parent pins, or if ``fetch_from`` is set and the output is not local.
-        TimeNetMissingParentError: If a parent or a payload source is missing and ``build_parents``
-            is not set.
+        LookupError: If no installed connector declares a dataset card for ``dataset_id``.
+        TimeNetInvalidCardError: If an installed connector's dataset card cannot be read.
+        TimeFValidationError: If parents cycle, a connector builds the wrong version, or fetching uses non-local output.
+        TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false.
         TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it.
-        TimeNetFileError: If a local output holds a damaged copy of a parent, and the plan cannot
-            fetch the parent again.
-        TimeNetAccessError: If the output is not local and the plan visits a release whose card, or
-            a card in its lineage, does not declare ``access: open``. Also if ``fetch_from`` is not
-            local and the plan would fetch such a release from it.
+        TimeNetFileError: If a local parent copy has a missing or resized file and cannot be fetched again.
+        TimeNetAccessError: If a release or its lineage lacks open access where a registry requires it.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk
     if fetch_from is not None and not isinstance(output, LocalRegistry):
         # A registry that is not local accepts a child only if it already holds the parents. Also, a
@@ -259,7 +236,7 @@ def plan_parents(
     # If the output already holds the requested release, a reader needs only its payload sources to
     # open it. Its declared parents matter only to a rebuild.
     held = None if force_build else _manifest(output, root)
-    if held is None and not isinstance(output, LocalRegistry):
+    if held is None and output.requires_open_access:
         require_open_access(card)
     searched = str(output) if fetch_from is None else f"{output} or {fetch_from}"
     graph = _Graph(output, fetch_from, build_parents=build_parents, searched=searched)
@@ -346,21 +323,16 @@ def _locate(
 ) -> tuple[Manifest | None, Literal["fetch"] | None]:
     # The damage is raised from a variable, the access error by require_open_access.
     # ruff: ignore[docstring-extraneous-exception]
-    """Find a release in the output, or else in ``fetch_from``.
+    """Find a release in ``output`` or ``fetch_from``.
 
-    The output must hold an intact copy. In a local output, a copy with a missing file or a file of
-    another size is damaged.
+    A local copy is intact when every file has its recorded size.
 
     Returns:
-        The manifest and ``None`` if the output holds an intact copy. The manifest and ``"fetch"`` if
-        the plan fetches the release. ``None`` and ``None`` if the plan builds it.
+        ``(manifest, None)`` for an output copy, ``(manifest, "fetch")`` for a fetch, or ``(None, None)`` if absent.
 
     Raises:
-        TimeNetFileError: If a local output holds a copy with a missing file or a file of another
-            size, and the plan cannot fetch the release again.
-        TimeNetAccessError: If the card of the release, or a card in its lineage, does not declare
-            ``access: open``, and the output that holds the release, or the ``fetch_from`` registry
-            to fetch it from, is not local.
+        TimeNetFileError: If the local copy has a missing or resized file and ``fetch_from`` lacks the release.
+        TimeNetAccessError: If the release or its lineage lacks open access where its registry requires it.
     """
     manifest = _manifest(output, reference)
     damage = None
@@ -375,14 +347,14 @@ def _locate(
         except TimeNetFileError as error:
             manifest, damage = None, error
     if manifest is not None:
-        if not isinstance(output, LocalRegistry):
+        if output.requires_open_access:
             require_open_access(manifest.metadata, manifest.lineage)
         return manifest, None
     fetched = None if fetch_from is None else _manifest(fetch_from, reference)
     if fetched is not None:
-        # A registry that is not local holds only releases whose cards declare access: open. So the
-        # plan refuses another release before any download.
-        if not isinstance(fetch_from, LocalRegistry):
+        # A registry that requires open access serves only releases whose cards declare access: open.
+        # So the plan refuses another release before any download.
+        if fetch_from is not None and fetch_from.requires_open_access:
             require_open_access(fetched.metadata, fetched.lineage)
         return fetched, "fetch"
     if damage is not None:
@@ -399,19 +371,19 @@ def _card_to_build(  # ruff: ignore[too-many-arguments]
     pinned: bool,
     searched: str,
 ) -> DatasetMetadata:
-    """Return the card to build a release that the output does not hold and the plan cannot fetch.
+    """Find a dataset card for building a missing release.
 
     Returns:
-        The card of the installed connector that builds exactly ``reference``.
+        The installed connector's card for the exact release in ``reference``.
 
     Raises:
-        TimeNetMissingParentError: If ``build_parents`` is not set.
-        TimeNetAccessError: If the output is not local and the card does not declare ``access: open``.
+        TimeNetMissingParentError: If ``build_parents`` is false.
+        TimeNetAccessError: If the card lacks open access and the output requires it.
     """  # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access
     if not build_parents:
-        raise TimeNetMissingParentError(reference, needed_by, pinned, searched)
+        raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned)
     card = _exact_card(reference)
-    if not isinstance(output, LocalRegistry):
+    if output.requires_open_access:
         require_open_access(card)
     return card
 

packages/timenet/src/timenet/engine/engine.py

@@ -113,34 +113,34 @@ def publish_pipeline(  # ruff: ignore[too-many-arguments]
     force: bool = False,
     pinned_by: Mapping[DatasetRef, PayloadSource] | None = None,
 ) -> str:
-    """Run a connector and publish its dataset to a registry.
+    """Build and publish a dataset to a writable 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.
+    Reuse an existing version unless ``force`` is set.
+    The dataset card and lineage must declare open access if the registry requires it.
 
     Args:
         connector: The connector to build.
-        registry: The destination registry.
-        cache_dir: Download directory, or ``None`` for ``<TIMENET_CACHE>/<dataset_id>``.
-        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.
+        registry: The writable registry to publish to.
+        cache_dir: The download directory, or ``None`` for ``<TIMENET_CACHE>/<dataset_id>``.
+        keep_cache: Keep the default cache after publishing. A supplied ``cache_dir`` is always kept.
+        progress_cb: Callback for writer progress events, or ``None``.
+        force: Publish again even if the version already exists.
+        pinned_by: Releases that read this release's files, mapped to file records the build must match.
 
     Returns:
-        The published version string.
+        The existing or newly published version string.
 
     Raises:
-        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.
+        TimeFValidationError: If a connector declares parents without overriding ``compose``, or changes its card.
+        TimeNetAccessError: If the card, dataset, or lineage lacks open access and the registry requires it.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access
     values_backend = _check_values_backend(connector)
     metadata = connector.metadata()
     dataset_id = metadata.dataset_id
     version = str(metadata.dataset_version)
     if not force and registry.exists(dataset_id, version):
         return version
-    if not isinstance(registry, LocalRegistry):
+    if registry.requires_open_access:
         require_open_access(metadata)
     _check_composes(connector, metadata)
 
@@ -149,7 +149,7 @@ def publish_pipeline(  # ruff: ignore[too-many-arguments]
 
     with open_parents(metadata, registry) as context:
         dataset = _compose(connector, metadata, cache, context)
-        if not isinstance(registry, LocalRegistry):
+        if registry.requires_open_access:
             require_open_access(dataset.metadata, dataset.lineage.values())
         dataset.derive_schema()
         registry.put_dataset(