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.

changes/4187.feature.md

@@ -0,0 +1,4 @@
+`ZipStore` now accepts an open binary file-like object in place of a path, enabling
+zip archives on remote storage (e.g. a file opened with `fsspec` or an
+`obstore.ReadableFile`). Operations that require a filesystem location
+(`clear`, `move`) raise `NotImplementedError` for file-object-backed stores.

docs/user-guide/storage.md

@@ -124,6 +124,19 @@ array = zarr.create_array(store=store, shape=(2,), dtype='float64')
 print(array)
 ```
 
+In place of a path, `ZipStore` also accepts an open binary file object (for
+example a file opened with `fsspec`, or an `obstore` reader), enabling zip
+archives on remote storage. The file must stay open for as long as the store
+is in use:
+
+```python exec="true" session="storage" source="above" result="ansi"
+store.close()
+f = open('data.zip', mode='rb')  # must stay open while the store is used
+array = zarr.open_array(store=zarr.storage.ZipStore(f), mode='r')
+print(array[:])
+f.close()
+```
+
 ### Remote Store
 
 The [`zarr.storage.FsspecStore`][] stores the contents of a Zarr hierarchy following the same

src/zarr/abc/url_pipeline.py

@@ -0,0 +1,174 @@
+"""
+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, ZarrFormat
+
+__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.
+    zarr_format : ZarrFormat | None
+        Zarr format selected by a format segment (`zarr2:`/`zarr3:`),
+        or None if unspecified.
+    """
+
+    store: Store
+    path: str = ""
+    zarr_format: ZarrFormat | None = None
+
+
+@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 explicit write modes (`"w"`, `"w-"`, `"a"`, `"r+"`)
+        and open read-only for `None` and `"r"`.
+    read_only : bool
+        True when `mode == "r"`. Adapters must construct their store
+        read-only when this is set.
+    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."""
+        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.
+        Adapters backed by their own I/O machinery (e.g. `icechunk:`)
+        should use [`preceding_url`][zarr.abc.url_pipeline.PipelineContext.preceding_url]
+        instead and never materialize the intermediate store.
+        """
+        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, an optional residual path within that store, and an
+        optional zarr format.
+
+        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

@@ -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,53 @@ 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.
+    """
+    from zarr.errors import URLPipelineError
+
+    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()