Skip to content

Commit 55cf114

Browse files
jhammanclaude
andcommitted
feat: url-pipeline core — parser, adapter ABC, registry, store hooks
Implements URL pipeline support (https://github.com/jbms/url-pipeline): '|'-chained URLs resolve through pluggable adapters registered under the 'zarr.url_adapters' entry-point group (entry-point name = URL scheme). - zarr.abc.url_pipeline: PipelineSegment, AdapterResolution, PipelineContext, URLPipelineAdapter (single-classmethod contract) - zarr.storage._url_pipeline: parse_pipeline / resolve_pipeline; the root sub-URL delegates to make_store so existing file/memory/fsspec routing is unchanged - registry: register_url_adapter / get_url_adapter / list_url_adapter_schemes (name check only; no adapter imports) - make_store/make_store_path route strings containing '|' (or a registered root scheme) through the resolver; residual store paths combine with the user-supplied path - StorePath gains a zarr_format attribute (populated by format segments in a follow-up) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 63fc294 commit 55cf114

15 files changed

Lines changed: 1133 additions & 3 deletions

File tree

changes/2943.feature.md

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

docs/api/zarr/abc/url_pipeline.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
title: url_pipeline
3+
---
4+
5+
::: zarr.abc.url_pipeline

mkdocs.yml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ nav:
4040
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.metadata</code>': api/zarr/abc/metadata.md
4141
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.numcodec</code>': api/zarr/abc/numcodec.md
4242
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.store</code>': api/zarr/abc/store.md
43+
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.url_pipeline</code>': api/zarr/abc/url_pipeline.md
4344
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.api</code>':
4445
- api/zarr/api/index.md
4546
- '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.api.asynchronous</code>': api/zarr/api/asynchronous.md

src/zarr/abc/url_pipeline.py

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
"""
2+
Abstract base class and data model for URL pipeline adapters.
3+
4+
A URL pipeline is a `|`-separated chain of sub-URLs, read outer-to-inner,
5+
as specified by https://github.com/jbms/url-pipeline. The first sub-URL (the
6+
*root*) locates a resource using a conventional URL, and each subsequent
7+
sub-URL names an *adapter* that reinterprets everything to its left:
8+
9+
s3://bucket/data.zip|zip:path/inside|zarr3:
10+
11+
Third-party packages provide adapters by subclassing
12+
[`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] and
13+
registering the class under the `zarr.url_adapters` entry-point group,
14+
using the URL scheme as the entry-point name.
15+
"""
16+
17+
from __future__ import annotations
18+
19+
from abc import ABC, abstractmethod
20+
from dataclasses import dataclass, field
21+
from typing import TYPE_CHECKING, Any
22+
23+
if TYPE_CHECKING:
24+
from collections.abc import Awaitable, Callable
25+
26+
from zarr.abc.store import Store
27+
from zarr.core.common import AccessModeLiteral
28+
29+
__all__ = [
30+
"AdapterResolution",
31+
"PipelineContext",
32+
"PipelineSegment",
33+
"URLPipelineAdapter",
34+
]
35+
36+
37+
@dataclass(frozen=True)
38+
class PipelineSegment:
39+
"""
40+
One `|`-delimited sub-URL of a URL pipeline.
41+
42+
Attributes
43+
----------
44+
scheme : str
45+
The lowercased URL scheme. Empty string only for a schemeless root
46+
(a bare local path).
47+
body : str
48+
The text after `scheme:` and before any `?`. Interpretation is
49+
scheme-defined; it is **not** URL-normalized, so case-significant
50+
content (e.g. icechunk snapshot IDs) is preserved.
51+
query : str | None
52+
The raw query string after `?`, or None. Interpretation is
53+
scheme-defined.
54+
raw : str
55+
The exact original sub-URL text, preserved for lossless
56+
reconstruction of the pipeline.
57+
"""
58+
59+
scheme: str
60+
body: str
61+
query: str | None
62+
raw: str
63+
64+
def __str__(self) -> str:
65+
return self.raw
66+
67+
68+
@dataclass(frozen=True)
69+
class AdapterResolution:
70+
"""
71+
The result of resolving a URL pipeline (or a prefix of one).
72+
73+
Attributes
74+
----------
75+
store : Store
76+
The resolved store.
77+
path : str
78+
Residual path *within* the store that the pipeline addresses
79+
(e.g. `"path/to/node"` for `...|icechunk://tag.v1/path/to/node`).
80+
Empty string when the pipeline addresses the store root.
81+
"""
82+
83+
store: Store
84+
path: str = ""
85+
86+
87+
@dataclass(frozen=True)
88+
class PipelineContext:
89+
"""
90+
Context handed to a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
91+
describing the pipeline to the left of its segment.
92+
93+
Attributes
94+
----------
95+
preceding : tuple[PipelineSegment, ...]
96+
The parsed sub-URLs to the left of the adapter's segment, outer to
97+
inner. Empty when the adapter's segment is the pipeline root.
98+
mode : AccessModeLiteral | None
99+
The access mode requested by the caller (e.g. `zarr.open(mode=...)`),
100+
or None when unspecified. Adapters for read-only resources should
101+
raise for unambiguous write modes (`"w"`, `"w-"`, `"r+"`) and
102+
open read-only otherwise. `"a"` (the `zarr.open` default) means
103+
open-or-create: read-only adapters serve the "open" half, and any
104+
subsequent write fails at the store level.
105+
read_only : bool
106+
True when the caller requires a read-only store (`mode == "r"`).
107+
Adapters must construct their store read-only when this is set;
108+
when it is False, they may construct a writable store if the
109+
underlying resource supports writing.
110+
storage_options : dict[str, Any] | None
111+
Options passed by the caller. By convention these configure the
112+
*root* sub-URL (e.g. fsspec options), but adapters may consume
113+
adapter-specific keys.
114+
"""
115+
116+
preceding: tuple[PipelineSegment, ...]
117+
mode: AccessModeLiteral | None
118+
read_only: bool
119+
storage_options: dict[str, Any] | None
120+
_resolver: Callable[[tuple[PipelineSegment, ...]], Awaitable[AdapterResolution]] = field(
121+
repr=False
122+
)
123+
124+
@property
125+
def preceding_url(self) -> str:
126+
"""
127+
The pipeline to the left of this segment, reconstructed exactly.
128+
129+
An adapter that consumes this string instead of calling
130+
[`resolve_preceding`][zarr.abc.url_pipeline.PipelineContext.resolve_preceding]
131+
takes ownership of the *entire* preceding pipeline: it must
132+
validate every preceding segment itself and raise
133+
[`URLPipelineError`][zarr.errors.URLPipelineError] for segments it
134+
does not understand, so that no segment is ever silently ignored.
135+
"""
136+
return "|".join(segment.raw for segment in self.preceding)
137+
138+
async def resolve_preceding(self) -> AdapterResolution:
139+
"""
140+
Resolve the preceding pipeline into a store.
141+
142+
This is the entry point for *wrapper* adapters (e.g. `zip:`) that
143+
operate on the resource produced by the segments to their left. It
144+
composes with any preceding adapters, because each segment is
145+
resolved by its own adapter. Adapters backed by their own I/O
146+
machinery (e.g. `icechunk:`) may instead consume
147+
[`preceding_url`][zarr.abc.url_pipeline.PipelineContext.preceding_url]
148+
and never materialize the intermediate store — subject to the
149+
ownership contract documented there.
150+
"""
151+
return await self._resolver(self.preceding)
152+
153+
154+
class URLPipelineAdapter(ABC):
155+
"""
156+
Handler for one URL pipeline scheme.
157+
158+
Subclasses implement a single classmethod,
159+
[`open_pipeline_segment`][zarr.abc.url_pipeline.URLPipelineAdapter.open_pipeline_segment],
160+
and are registered under the `zarr.url_adapters` entry-point group with
161+
the URL scheme as the entry-point name:
162+
163+
[project.entry-points."zarr.url_adapters"]
164+
myscheme = "mypackage.zarr_adapter:MyAdapter"
165+
166+
An adapter is used in two positions:
167+
168+
- as an *adapter segment*: `s3://bucket/repo|icechunk://tag.v1` — the
169+
context carries the preceding sub-URLs;
170+
- as a *root scheme*: `gh://org/repo` — `context.preceding` is empty.
171+
"""
172+
173+
@classmethod
174+
@abstractmethod
175+
async def open_pipeline_segment(
176+
cls, segment: PipelineSegment, context: PipelineContext
177+
) -> AdapterResolution:
178+
"""
179+
Resolve `segment` (in the context of the pipeline to its left)
180+
into a store and an optional residual path within that store.
181+
182+
The returned store must already be open and must honor
183+
`context.read_only`.
184+
"""
185+
...

src/zarr/errors.py

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
"MetadataValidationError",
1313
"NegativeStepError",
1414
"NodeTypeValidationError",
15+
"URLPipelineError",
1516
"UnstableSpecificationWarning",
1617
"VindexInvalidSelectionError",
1718
"ZarrDeprecationWarning",
@@ -100,6 +101,12 @@ class UnknownCodecError(BaseZarrError):
100101
"""
101102

102103

104+
class URLPipelineError(BaseZarrError):
105+
"""
106+
Raised when a URL pipeline cannot be parsed or resolved.
107+
"""
108+
109+
103110
class NodeTypeValidationError(MetadataValidationError):
104111
"""
105112
Specialized exception when the node_type of the metadata document is incorrect.

src/zarr/registry.py

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77

88
from zarr.core.config import BadConfigError, config
99
from zarr.core.dtype import data_type_registry
10-
from zarr.errors import ZarrUserWarning
10+
from zarr.errors import URLPipelineError, ZarrUserWarning
1111

1212
if TYPE_CHECKING:
1313
from importlib.metadata import EntryPoint
@@ -21,6 +21,7 @@
2121
CodecPipeline,
2222
)
2323
from zarr.abc.numcodec import Numcodec
24+
from zarr.abc.url_pipeline import URLPipelineAdapter
2425
from zarr.core.buffer import Buffer, NDBuffer
2526
from zarr.core.chunk_key_encodings import ChunkKeyEncoding
2627
from zarr.core.common import JSON
@@ -32,11 +33,14 @@
3233
"get_codec_class",
3334
"get_ndbuffer_class",
3435
"get_pipeline_class",
36+
"get_url_adapter",
37+
"list_url_adapter_schemes",
3538
"register_buffer",
3639
"register_chunk_key_encoding",
3740
"register_codec",
3841
"register_ndbuffer",
3942
"register_pipeline",
43+
"register_url_adapter",
4044
]
4145

4246

@@ -62,6 +66,7 @@ def register(self, cls: type[T], qualname: str | None = None) -> None:
6266
_buffer_registry: Registry[Buffer] = Registry()
6367
_ndbuffer_registry: Registry[NDBuffer] = Registry()
6468
_chunk_key_encoding_registry: Registry[ChunkKeyEncoding] = Registry()
69+
_url_adapter_registry: Registry[URLPipelineAdapter] = Registry()
6570

6671
"""
6772
The registry module is responsible for managing implementations of codecs,
@@ -108,6 +113,8 @@ def _collect_entrypoints() -> list[Registry[Any]]:
108113
entry_points.select(group="zarr", name="chunk_key_encoding")
109114
)
110115

116+
_url_adapter_registry.lazy_load_list.extend(entry_points.select(group="zarr.url_adapters"))
117+
111118
_pipeline_registry.lazy_load_list.extend(entry_points.select(group="zarr.codec_pipeline"))
112119
_pipeline_registry.lazy_load_list.extend(
113120
entry_points.select(group="zarr", name="codec_pipeline")
@@ -124,6 +131,7 @@ def _collect_entrypoints() -> list[Registry[Any]]:
124131
_buffer_registry,
125132
_ndbuffer_registry,
126133
_chunk_key_encoding_registry,
134+
_url_adapter_registry,
127135
]
128136

129137

@@ -303,6 +311,51 @@ def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]:
303311
return _chunk_key_encoding_registry[key]
304312

305313

314+
def register_url_adapter(scheme: str, cls: type[URLPipelineAdapter]) -> None:
315+
"""
316+
Register a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
317+
class for a URL scheme.
318+
"""
319+
_url_adapter_registry.register(cls, scheme.lower())
320+
321+
322+
def list_url_adapter_schemes() -> set[str]:
323+
"""
324+
The set of URL schemes with a registered URL pipeline adapter.
325+
326+
Includes adapters advertised via not-yet-loaded `zarr.url_adapters`
327+
entry points; consulting this does not import any adapter code.
328+
"""
329+
return set(_url_adapter_registry) | {e.name for e in _url_adapter_registry.lazy_load_list}
330+
331+
332+
def get_url_adapter(scheme: str) -> type[URLPipelineAdapter]:
333+
"""
334+
Get the URL pipeline adapter class registered for `scheme`.
335+
336+
Loads pending `zarr.url_adapters` entry points for this scheme only, so
337+
resolving one scheme never imports other providers' packages.
338+
"""
339+
key = scheme.lower()
340+
if key not in _url_adapter_registry:
341+
remaining = []
342+
for entry_point in _url_adapter_registry.lazy_load_list:
343+
if entry_point.name == key:
344+
_url_adapter_registry.register(entry_point.load(), qualname=key)
345+
else:
346+
remaining.append(entry_point)
347+
_url_adapter_registry.lazy_load_list[:] = remaining
348+
try:
349+
return _url_adapter_registry[key]
350+
except KeyError:
351+
registered = sorted(list_url_adapter_schemes())
352+
raise URLPipelineError(
353+
f"no URL pipeline adapter is registered for scheme {scheme!r}. "
354+
f"Registered schemes: {registered}. Adapters are provided by "
355+
"packages via the 'zarr.url_adapters' entry-point group."
356+
) from None
357+
358+
306359
_collect_entrypoints()
307360

308361

src/zarr/storage/_common.py

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,15 @@
2121
AccessModeLiteral,
2222
ZarrFormat,
2323
)
24-
from zarr.errors import ContainsArrayAndGroupError, ContainsArrayError, ContainsGroupError
24+
from zarr.errors import (
25+
ContainsArrayAndGroupError,
26+
ContainsArrayError,
27+
ContainsGroupError,
28+
URLPipelineError,
29+
)
2530
from zarr.storage._local import LocalStore
2631
from zarr.storage._memory import ManagedMemoryStore, MemoryStore
32+
from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline
2733
from zarr.storage._utils import _join_paths, normalize_path, parse_store_url
2834

2935
_has_fsspec = importlib.util.find_spec("fsspec")
@@ -348,6 +354,15 @@ async def make_store(
348354
"""
349355
from zarr.storage._fsspec import FsspecStore # circular import
350356

357+
if isinstance(store_like, str) and is_url_pipeline(store_like):
358+
result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
359+
if result.path:
360+
raise URLPipelineError(
361+
f"the URL pipeline {store_like!r} resolves to a path inside a store; "
362+
"use zarr.open() or make_store_path() instead of make_store()"
363+
)
364+
return result.store
365+
351366
# Parse URL early so we can reuse the result for both validation and routing
352367
parsed = parse_store_url(store_like) if isinstance(store_like, str) else None
353368

@@ -453,6 +468,17 @@ async def make_store_path(
453468
"""
454469
path_normalized = normalize_path(path)
455470

471+
if isinstance(store_like, str) and is_url_pipeline(store_like):
472+
result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
473+
combined_path = _join_paths([normalize_path(result.path), path_normalized])
474+
# mode "a" (the zarr.open default) means open-or-create; when the
475+
# pipeline resolved to a read-only resource, honor the "open" half
476+
# rather than failing outright. Writes still fail at the store level.
477+
open_mode: AccessModeLiteral | None = (
478+
"r" if (mode == "a" and result.store.read_only) else mode
479+
)
480+
return await StorePath.open(result.store, path=combined_path, mode=open_mode)
481+
456482
if isinstance(store_like, StorePath):
457483
# Already a StorePath
458484
if storage_options:

0 commit comments

Comments
 (0)