feat: url-pipeline core — parser, adapter ABC, registry, store hooks

changes/2943.feature.md

@@ -0,0 +1,9 @@
+Added core support for URL pipelines (https://github.com/jbms/url-pipeline):
+`|`-chained URLs that address zarr data through nested storage layers, e.g.
+`s3://bucket/data.zip|zip:|zarr3:`. This PR adds the parser, the single-method
+`zarr.abc.url_pipeline.URLPipelineAdapter` interface, and the
+`zarr.url_adapters` entry-point group through which third-party packages
+(e.g. Icechunk) register adapters for their own schemes. Adapters for a scheme
+are loaded lazily and individually; URLs without a `|` separator (and without
+a registered root scheme) are handled exactly as before. Builtin adapters
+(`zip:`, `zarr2:`/`zarr3:`) follow in separate pull requests.

src/zarr/abc/url_pipeline.py

@@ -0,0 +1,185 @@
+"""
+Abstract base class and data model for URL pipeline adapters.
+
+A URL pipeline is a `|`-separated chain of sub-URLs, read outer-to-inner,
+as specified by https://github.com/jbms/url-pipeline. The first sub-URL (the
+*root*) locates a resource using a conventional URL, and each subsequent
+sub-URL names an *adapter* that reinterprets everything to its left:
+
+    s3://bucket/data.zip|zip:path/inside|zarr3:
+
+Third-party packages provide adapters by subclassing
+[`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] and
+registering the class under the `zarr.url_adapters` entry-point group,
+using the URL scheme as the entry-point name.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+    from zarr.abc.store import Store
+    from zarr.core.common import AccessModeLiteral
+
+__all__ = [
+    "AdapterResolution",
+    "PipelineContext",
+    "PipelineSegment",
+    "URLPipelineAdapter",
+]
+
+
+@dataclass(frozen=True)
+class PipelineSegment:
+    """
+    One `|`-delimited sub-URL of a URL pipeline.
+
+    Attributes
+    ----------
+    scheme : str
+        The lowercased URL scheme. Empty string only for a schemeless root
+        (a bare local path).
+    body : str
+        The text after `scheme:` and before any `?`. Interpretation is
+        scheme-defined; it is **not** URL-normalized, so case-significant
+        content (e.g. icechunk snapshot IDs) is preserved.
+    query : str | None
+        The raw query string after `?`, or None. Interpretation is
+        scheme-defined.
+    raw : str
+        The exact original sub-URL text, preserved for lossless
+        reconstruction of the pipeline.
+    """
+
+    scheme: str
+    body: str
+    query: str | None
+    raw: str
+
+    def __str__(self) -> str:
+        return self.raw
+
+
+@dataclass(frozen=True)
+class AdapterResolution:
+    """
+    The result of resolving a URL pipeline (or a prefix of one).
+
+    Attributes
+    ----------
+    store : Store
+        The resolved store.
+    path : str
+        Residual path *within* the store that the pipeline addresses
+        (e.g. `"path/to/node"` for `...|icechunk://tag.v1/path/to/node`).
+        Empty string when the pipeline addresses the store root.
+    """
+
+    store: Store
+    path: str = ""
+
+
+@dataclass(frozen=True)
+class PipelineContext:
+    """
+    Context handed to a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
+    describing the pipeline to the left of its segment.
+
+    Attributes
+    ----------
+    preceding : tuple[PipelineSegment, ...]
+        The parsed sub-URLs to the left of the adapter's segment, outer to
+        inner. Empty when the adapter's segment is the pipeline root.
+    mode : AccessModeLiteral | None
+        The access mode requested by the caller (e.g. `zarr.open(mode=...)`),
+        or None when unspecified. Adapters for read-only resources should
+        raise for unambiguous write modes (`"w"`, `"w-"`, `"r+"`) and
+        open read-only otherwise. `"a"` (the `zarr.open` default) means
+        open-or-create: read-only adapters serve the "open" half, and any
+        subsequent write fails at the store level.
+    read_only : bool
+        True when the caller requires a read-only store (`mode == "r"`).
+        Adapters must construct their store read-only when this is set;
+        when it is False, they may construct a writable store if the
+        underlying resource supports writing.
+    storage_options : dict[str, Any] | None
+        Options passed by the caller. By convention these configure the
+        *root* sub-URL (e.g. fsspec options), but adapters may consume
+        adapter-specific keys.
+    """
+
+    preceding: tuple[PipelineSegment, ...]
+    mode: AccessModeLiteral | None
+    read_only: bool
+    storage_options: dict[str, Any] | None
+    _resolver: Callable[[tuple[PipelineSegment, ...]], Awaitable[AdapterResolution]] = field(
+        repr=False
+    )
+
+    @property
+    def preceding_url(self) -> str:
+        """
+        The pipeline to the left of this segment, reconstructed exactly.
+
+        An adapter that consumes this string instead of calling
+        [`resolve_preceding`][zarr.abc.url_pipeline.PipelineContext.resolve_preceding]
+        takes ownership of the *entire* preceding pipeline: it must
+        validate every preceding segment itself and raise
+        [`URLPipelineError`][zarr.errors.URLPipelineError] for segments it
+        does not understand, so that no segment is ever silently ignored.
+        """
+        return "|".join(segment.raw for segment in self.preceding)
+
+    async def resolve_preceding(self) -> AdapterResolution:
+        """
+        Resolve the preceding pipeline into a store.
+
+        This is the entry point for *wrapper* adapters (e.g. `zip:`) that
+        operate on the resource produced by the segments to their left. It
+        composes with any preceding adapters, because each segment is
+        resolved by its own adapter. Adapters backed by their own I/O
+        machinery (e.g. `icechunk:`) may instead consume
+        [`preceding_url`][zarr.abc.url_pipeline.PipelineContext.preceding_url]
+        and never materialize the intermediate store — subject to the
+        ownership contract documented there.
+        """
+        return await self._resolver(self.preceding)
+
+
+class URLPipelineAdapter(ABC):
+    """
+    Handler for one URL pipeline scheme.
+
+    Subclasses implement a single classmethod,
+    [`open_pipeline_segment`][zarr.abc.url_pipeline.URLPipelineAdapter.open_pipeline_segment],
+    and are registered under the `zarr.url_adapters` entry-point group with
+    the URL scheme as the entry-point name:
+
+        [project.entry-points."zarr.url_adapters"]
+        myscheme = "mypackage.zarr_adapter:MyAdapter"
+
+    An adapter is used in two positions:
+
+    - as an *adapter segment*: `s3://bucket/repo|icechunk://tag.v1` — the
+      context carries the preceding sub-URLs;
+    - as a *root scheme*: `gh://org/repo` — `context.preceding` is empty.
+    """
+
+    @classmethod
+    @abstractmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        """
+        Resolve `segment` (in the context of the pipeline to its left)
+        into a store and an optional residual path within that store.
+
+        The returned store must already be open and must honor
+        `context.read_only`.
+        """
+        ...

src/zarr/errors.py

@@ -12,6 +12,7 @@
     "MetadataValidationError",
     "NegativeStepError",
     "NodeTypeValidationError",
+    "URLPipelineError",
     "UnstableSpecificationWarning",
     "VindexInvalidSelectionError",
     "ZarrDeprecationWarning",
@@ -100,6 +101,12 @@ class UnknownCodecError(BaseZarrError):
     """
 
 
+class URLPipelineError(BaseZarrError):
+    """
+    Raised when a URL pipeline cannot be parsed or resolved.
+    """
+
+
 class NodeTypeValidationError(MetadataValidationError):
     """
     Specialized exception when the node_type of the metadata document is incorrect.

src/zarr/registry.py

@@ -7,7 +7,7 @@
 
 from zarr.core.config import BadConfigError, config
 from zarr.core.dtype import data_type_registry
-from zarr.errors import ZarrUserWarning
+from zarr.errors import URLPipelineError, ZarrUserWarning
 
 if TYPE_CHECKING:
     from importlib.metadata import EntryPoint
@@ -21,6 +21,7 @@
         CodecPipeline,
     )
     from zarr.abc.numcodec import Numcodec
+    from zarr.abc.url_pipeline import URLPipelineAdapter
     from zarr.core.buffer import Buffer, NDBuffer
     from zarr.core.chunk_key_encodings import ChunkKeyEncoding
     from zarr.core.common import JSON
@@ -32,11 +33,14 @@
     "get_codec_class",
     "get_ndbuffer_class",
     "get_pipeline_class",
+    "get_url_adapter",
+    "list_url_adapter_schemes",
     "register_buffer",
     "register_chunk_key_encoding",
     "register_codec",
     "register_ndbuffer",
     "register_pipeline",
+    "register_url_adapter",
 ]
 
 
@@ -62,6 +66,7 @@ def register(self, cls: type[T], qualname: str | None = None) -> None:
 _buffer_registry: Registry[Buffer] = Registry()
 _ndbuffer_registry: Registry[NDBuffer] = Registry()
 _chunk_key_encoding_registry: Registry[ChunkKeyEncoding] = Registry()
+_url_adapter_registry: Registry[URLPipelineAdapter] = Registry()
 
 """
 The registry module is responsible for managing implementations of codecs,
@@ -108,6 +113,8 @@ def _collect_entrypoints() -> list[Registry[Any]]:
         entry_points.select(group="zarr", name="chunk_key_encoding")
     )
 
+    _url_adapter_registry.lazy_load_list.extend(entry_points.select(group="zarr.url_adapters"))
+
     _pipeline_registry.lazy_load_list.extend(entry_points.select(group="zarr.codec_pipeline"))
     _pipeline_registry.lazy_load_list.extend(
         entry_points.select(group="zarr", name="codec_pipeline")
@@ -124,6 +131,7 @@ def _collect_entrypoints() -> list[Registry[Any]]:
         _buffer_registry,
         _ndbuffer_registry,
         _chunk_key_encoding_registry,
+        _url_adapter_registry,
     ]
 
 
@@ -303,6 +311,51 @@ def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]:
     return _chunk_key_encoding_registry[key]
 
 
+def register_url_adapter(scheme: str, cls: type[URLPipelineAdapter]) -> None:
+    """
+    Register a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
+    class for a URL scheme.
+    """
+    _url_adapter_registry.register(cls, scheme.lower())
+
+
+def list_url_adapter_schemes() -> set[str]:
+    """
+    The set of URL schemes with a registered URL pipeline adapter.
+
+    Includes adapters advertised via not-yet-loaded `zarr.url_adapters`
+    entry points; consulting this does not import any adapter code.
+    """
+    return set(_url_adapter_registry) | {e.name for e in _url_adapter_registry.lazy_load_list}
+
+
+def get_url_adapter(scheme: str) -> type[URLPipelineAdapter]:
+    """
+    Get the URL pipeline adapter class registered for `scheme`.
+
+    Loads pending `zarr.url_adapters` entry points for this scheme only, so
+    resolving one scheme never imports other providers' packages.
+    """
+    key = scheme.lower()
+    if key not in _url_adapter_registry:
+        remaining = []
+        for entry_point in _url_adapter_registry.lazy_load_list:
+            if entry_point.name == key:
+                _url_adapter_registry.register(entry_point.load(), qualname=key)
+            else:
+                remaining.append(entry_point)
+        _url_adapter_registry.lazy_load_list[:] = remaining
+    try:
+        return _url_adapter_registry[key]
+    except KeyError:
+        registered = sorted(list_url_adapter_schemes())
+        raise URLPipelineError(
+            f"no URL pipeline adapter is registered for scheme {scheme!r}. "
+            f"Registered schemes: {registered}. Adapters are provided by "
+            "packages via the 'zarr.url_adapters' entry-point group."
+        ) from None
+
+
 _collect_entrypoints()
 
 

src/zarr/storage/_common.py

@@ -21,9 +21,15 @@
     AccessModeLiteral,
     ZarrFormat,
 )
-from zarr.errors import ContainsArrayAndGroupError, ContainsArrayError, ContainsGroupError
+from zarr.errors import (
+    ContainsArrayAndGroupError,
+    ContainsArrayError,
+    ContainsGroupError,
+    URLPipelineError,
+)
 from zarr.storage._local import LocalStore
 from zarr.storage._memory import ManagedMemoryStore, MemoryStore
+from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline
 from zarr.storage._utils import _join_paths, normalize_path, parse_store_url
 
 _has_fsspec = importlib.util.find_spec("fsspec")
@@ -348,6 +354,15 @@ async def make_store(
     """
     from zarr.storage._fsspec import FsspecStore  # circular import
 
+    if isinstance(store_like, str) and is_url_pipeline(store_like):
+        result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
+        if result.path:
+            raise URLPipelineError(
+                f"the URL pipeline {store_like!r} resolves to a path inside a store; "
+                "use zarr.open() or make_store_path() instead of make_store()"
+            )
+        return result.store
+
     # Parse URL early so we can reuse the result for both validation and routing
     parsed = parse_store_url(store_like) if isinstance(store_like, str) else None
 
@@ -453,6 +468,17 @@ async def make_store_path(
     """
     path_normalized = normalize_path(path)
 
+    if isinstance(store_like, str) and is_url_pipeline(store_like):
+        result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
+        combined_path = _join_paths([normalize_path(result.path), path_normalized])
+        # mode "a" (the zarr.open default) means open-or-create; when the
+        # pipeline resolved to a read-only resource, honor the "open" half
+        # rather than failing outright. Writes still fail at the store level.
+        open_mode: AccessModeLiteral | None = (
+            "r" if (mode == "a" and result.store.read_only) else mode
+        )
+        return await StorePath.open(result.store, path=combined_path, mode=open_mode)
+
     if isinstance(store_like, StorePath):
         # Already a StorePath
         if storage_options: