diff --git a/changes/4192.feature.md b/changes/4192.feature.md new file mode 100644 index 0000000000..757aa13895 --- /dev/null +++ b/changes/4192.feature.md @@ -0,0 +1,25 @@ +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. Builtin adapters (`zip:`, +`zarr2:`/`zarr3:`) follow in separate pull requests. + +Behavior notes: + +- The `|` character is now reserved as the pipeline delimiter in every string + store specification, and no percent-escape is decoded; pass a `pathlib.Path` + to address a local file whose name contains `|`. URLs without a `|` (and + without a registered root adapter scheme) are handled exactly as before — + registered adapters cannot intercept zarr's native `file:`/`memory:` + routing, and fsspec chained URLs (`zip::s3://...`) keep flowing to fsspec. +- Inside a pipeline, `memory:` and `file:` roots follow the URL pipeline + spec's semantics (spelling equivalences; `file:` must be absolute, with at + most a `localhost` authority). +- Mode `"a"` (open-or-create, the `zarr.open` default) on a *read-only* store + now serves the "open" half instead of raising upfront, for all stores; + unambiguous write modes (`"w"`, `"w-"`, `"r+"`) still raise. +- For root-adapter URLs (e.g. `gh://org/repo`), `storage_options` are handed + to the adapter and are not validated as used by `make_store`. diff --git a/docs/api/zarr/abc/index.md b/docs/api/zarr/abc/index.md index 7e15cb2a51..d155f60a5a 100644 --- a/docs/api/zarr/abc/index.md +++ b/docs/api/zarr/abc/index.md @@ -11,3 +11,4 @@ Abstract base classes for extending Zarr-Python. - **[zarr.abc.metadata](./metadata.md)** - Creating metadata classes compatible with the Zarr API - **[zarr.abc.numcodec](./numcodec.md)** - Protocols and classes for modeling codec interface used by numcodecs - **[zarr.abc.store](./store.md)** - ABC for implementing Zarr stores and managing getting and setting bytes in a store +- **[zarr.abc.url_pipeline](./url_pipeline.md)** - ABC for implementing [URL pipeline](https://github.com/jbms/url-pipeline) adapters diff --git a/docs/api/zarr/abc/url_pipeline.md b/docs/api/zarr/abc/url_pipeline.md new file mode 100644 index 0000000000..2606d4d5d4 --- /dev/null +++ b/docs/api/zarr/abc/url_pipeline.md @@ -0,0 +1,5 @@ +--- +title: url_pipeline +--- + +::: zarr.abc.url_pipeline diff --git a/docs/user-guide/storage.md b/docs/user-guide/storage.md index a34e2e2874..0b86bef2b0 100644 --- a/docs/user-guide/storage.md +++ b/docs/user-guide/storage.md @@ -103,6 +103,31 @@ print(group) - a [`Store`][zarr.abc.store.Store] or [`StorePath`][zarr.storage.StorePath] - see explicit store creation below. +## URL Pipelines {#user-guide-url-pipelines} + +Zarr supports [URL pipelines](https://github.com/jbms/url-pipeline): `|`-chained URLs +that address zarr data through nested storage layers, read left to right. The first +sub-URL locates a resource with a conventional URL; each subsequent sub-URL names an +*adapter* that reinterprets everything to its left (e.g. +`s3://bucket/data.zip|zip:|zarr3:`). Adapters are provided by packages through the +`zarr.url_adapters` entry-point group — see +[`zarr.abc.url_pipeline`][zarr.abc.url_pipeline] for the adapter interface. Builtin +adapters (`zip:`, `zarr2:`/`zarr3:`) are under development and will expand this +section. URLs without a `|` (and without a registered root scheme) are handled +exactly as before. + +`storage_options` passed to `zarr.open` apply to the *root* sub-URL (e.g. fsspec +options for `s3://...`); adapters may consume adapter-specific, namespaced keys. +Non-dict forms of `storage_options` are reserved for future per-segment +configuration. + +The `|` character is reserved as the pipeline delimiter in every string store +specification, and no percent-escape is decoded: to address a local file whose +*name* contains `|` (or `#`), pass a `pathlib.Path` instead of a string. +Registered adapters cannot intercept zarr's native `file:` and `memory:` root +schemes, and fsspec's chained-URL syntax (`zip::s3://...`) keeps flowing to +fsspec. + ## Explicit Store Creation In some cases, it may be helpful to create a store instance directly. Zarr-Python offers diff --git a/mkdocs.yml b/mkdocs.yml index 4d06701a87..58b5d3344a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -43,6 +43,7 @@ nav: - ' zarr.abc.metadata': api/zarr/abc/metadata.md - ' zarr.abc.numcodec': api/zarr/abc/numcodec.md - ' zarr.abc.store': api/zarr/abc/store.md + - ' zarr.abc.url_pipeline': api/zarr/abc/url_pipeline.md - ' zarr.api': - api/zarr/api/index.md - ' zarr.api.asynchronous': api/zarr/api/asynchronous.md diff --git a/src/zarr/abc/url_pipeline.py b/src/zarr/abc/url_pipeline.py new file mode 100644 index 0000000000..0e0363bcf6 --- /dev/null +++ b/src/zarr/abc/url_pipeline.py @@ -0,0 +1,256 @@ +""" +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 + +import enum +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from zarr.errors import URLPipelineError + +if TYPE_CHECKING: + from zarr.abc.store import Store + from zarr.core.common import AccessModeLiteral, ZarrFormat + +__all__ = [ + "AdapterResolution", + "PipelineContext", + "PipelineSegment", + "URLPipelineAdapter", +] + + +class _Unset(enum.Enum): + token = 0 + + +_UNSET = _Unset.token + + +@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), which is treated as opaque text. + 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. A wrapper adapter that re-wraps a preceding + resolution must carry every field it does not change forward — + prefer `dataclasses.replace(preceding, store=..., path=...)` over + reconstructing, so fields added later are never silently dropped. + """ + + 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 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. + storage_options : dict[str, Any] | None + Options passed by the caller. By convention these configure the + *root* sub-URL (e.g. fsspec options); adapters may consume + adapter-specific keys, and should namespace them (e.g. + `myscheme_credentials`) to avoid collisions with other segments' + backends. An adapter that consumes keys should strip them before + resolving the rest of the pipeline, by passing the reduced mapping + to [`resolve_preceding`][zarr.abc.url_pipeline.PipelineContext.resolve_preceding]. + Non-dict forms of the caller-facing `storage_options` argument are + reserved for future per-segment configuration (one mapping per + pipeline segment); this attribute will remain a single mapping — + the one addressed to this adapter's segment. + """ + + preceding: tuple[PipelineSegment, ...] + mode: AccessModeLiteral | None + storage_options: dict[str, Any] | None + + @property + def read_only(self) -> bool: + """ + True when the caller requires a read-only store (`mode == "r"`). + + Adapters must construct their store read-only when this is set + (the resolver enforces it afterwards); when it is False, they may + construct a writable store if the underlying resource supports + writing. + """ + return self.mode == "r" + + @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, + *, + mode: AccessModeLiteral | _Unset | None = _UNSET, + storage_options: dict[str, Any] | _Unset | None = _UNSET, + ) -> 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. + + Parameters + ---------- + mode : AccessModeLiteral | None, optional + Override the mode used to resolve the preceding pipeline. + Wrapper adapters that only read the preceding resource should + pass `mode="r"` so the root is opened read-only and without + create-on-open side effects, regardless of the caller's mode. + When omitted, the caller's mode is used. + storage_options : dict | None, optional + Override the options forwarded to the preceding pipeline. An + adapter that consumed adapter-specific keys should pass the + remaining mapping here (or None when nothing remains), so the + root store never sees keys that were not addressed to it. + When omitted, the caller's options are forwarded unchanged. + """ + from zarr.storage._url_pipeline import _resolve + + if not self.preceding: + raise URLPipelineError( + "this adapter segment is at the pipeline root; " + "there is no preceding sub-URL to resolve" + ) + return await _resolve( + self.preceding, + mode=self.mode if isinstance(mode, _Unset) else mode, + storage_options=( + self.storage_options if isinstance(storage_options, _Unset) else storage_options + ), + ) + + +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"] + mypackage.myscheme = "mypackage.zarr_adapter:MyAdapter" + + Nonstandard schemes should be vendor-prefixed (`vendor.scheme`) per the + URL pipeline specification. + + 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` (the resolver additionally enforces it by + downgrading — or rejecting — a writable store when the caller + required read-only). + + This coroutine runs on zarr's internal I/O event loop. It must not + block (do I/O through async APIs or a thread executor) and must not + call zarr's synchronous API (`zarr.open`, `Group.open`, or anything + else that uses `zarr.core.sync.sync`) — doing so raises + `SyncError`. To open the preceding pipeline, use + [`resolve_preceding`][zarr.abc.url_pipeline.PipelineContext.resolve_preceding]. + """ + ... diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py index 3bdc254ea5..2ae05bef6f 100644 --- a/src/zarr/api/asynchronous.py +++ b/src/zarr/api/asynchronous.py @@ -108,6 +108,24 @@ def _infer_overwrite(mode: AccessModeLiteral) -> bool: return mode in _OVERWRITE_MODES +def _merge_pipeline_zarr_format( + store_path: StorePath, zarr_format: ZarrFormat | None +) -> ZarrFormat | None: + """ + Combine a zarr format selected by a URL pipeline segment (`zarr2:` / + `zarr3:`) with the caller's `zarr_format` argument. Explicitly + conflicting selections raise. + """ + if store_path.zarr_format is None: + return zarr_format + if zarr_format is not None and zarr_format != store_path.zarr_format: + raise ValueError( + f"zarr_format={zarr_format} conflicts with the 'zarr{store_path.zarr_format}:' " + "segment of the URL pipeline" + ) + return store_path.zarr_format + + def _warn_unimplemented_kwargs(kwargs: dict[str, Any]) -> None: """ Emit a "not yet implemented" warning for each provided keyword argument that is not None. @@ -389,6 +407,7 @@ async def open( else: mode = "a" store_path = await make_store_path(store, mode=mode, path=path, storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) # TODO: the mode check below seems wrong! if "shape" not in kwargs and mode in {"a", "r", "r+", "w"}: @@ -494,13 +513,14 @@ async def save_array( **kwargs Passed through to [`create`][zarr.api.asynchronous.create], e.g., compressor. """ - if zarr_format is None: - zarr_format = _default_zarr_format() if not isinstance(arr, NDArrayLike): raise TypeError("arr argument must be numpy or other NDArrayLike array") mode = kwargs.pop("mode", "a") store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) + if zarr_format is None: + zarr_format = _default_zarr_format() if np.isscalar(arr): arr = np.array(arr) shape = arr.shape @@ -550,6 +570,7 @@ async def save_group( """ store_path = await make_store_path(store, path=path, mode="w", storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) if zarr_format is None: zarr_format = _default_zarr_format() @@ -762,12 +783,12 @@ async def create_group( The new group. """ - if zarr_format is None: - zarr_format = _default_zarr_format() - mode: Literal["a"] = "a" store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) + if zarr_format is None: + zarr_format = _default_zarr_format() return await AsyncGroup.from_store( store=store_path, @@ -857,6 +878,7 @@ async def open_group( ) store_path = await make_store_path(store, mode=mode, storage_options=storage_options, path=path) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) if attributes is None: attributes = {} @@ -1041,9 +1063,6 @@ async def create( z : array The array. """ - if zarr_format is None: - zarr_format = _default_zarr_format() - _warn_unimplemented_kwargs( { "synchronizer": synchronizer, @@ -1063,6 +1082,9 @@ async def create( if mode is None: mode = "a" store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) + if zarr_format is None: + zarr_format = _default_zarr_format() config_parsed = parse_array_config(config) @@ -1262,6 +1284,7 @@ async def open_array( mode = kwargs.pop("mode", None) store_path = await make_store_path(store, path=path, mode=mode, storage_options=storage_options) + zarr_format = _merge_pipeline_zarr_format(store_path, zarr_format) if "write_empty_chunks" in kwargs: _warn_write_empty_chunks_kwarg() diff --git a/src/zarr/errors.py b/src/zarr/errors.py index 781bebe534..de04a5bdaa 100644 --- a/src/zarr/errors.py +++ b/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. diff --git a/src/zarr/registry.py b/src/zarr/registry.py index c2c0eb2921..53875c015c 100644 --- a/src/zarr/registry.py +++ b/src/zarr/registry.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading import warnings from collections import defaultdict from importlib.metadata import entry_points as get_entry_points @@ -7,7 +8,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 +22,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 +34,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 +67,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 +114,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 +132,7 @@ def _collect_entrypoints() -> list[Registry[Any]]: _buffer_registry, _ndbuffer_registry, _chunk_key_encoding_registry, + _url_adapter_registry, ] @@ -303,6 +312,72 @@ 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. + + Registering a scheme that already has an adapter replaces it and emits a + [`ZarrUserWarning`][zarr.errors.ZarrUserWarning]. + """ + key = scheme.lower() + previous = _url_adapter_registry.get(key) + if previous is not None and previous is not cls: + warnings.warn( + f"URL pipeline adapter for scheme {scheme!r} is being replaced: " + f"{fully_qualified_name(previous)} -> {fully_qualified_name(cls)}", + category=ZarrUserWarning, + stacklevel=2, + ) + _url_adapter_registry.register(cls, key) + + +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. + Schemes are case-insensitive and reported lowercased. + """ + return set(_url_adapter_registry) | { + e.name.lower() for e in _url_adapter_registry.lazy_load_list + } + + +_url_adapter_lock = threading.Lock() + + +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() + # The lock keeps concurrent first-time resolutions of different schemes + # from clobbering each other's rebuild of the pending entry-point list. + with _url_adapter_lock: + if key not in _url_adapter_registry: + remaining = [] + for entry_point in _url_adapter_registry.lazy_load_list: + if entry_point.name.lower() == 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() diff --git a/src/zarr/storage/__init__.py b/src/zarr/storage/__init__.py index f1bd1724af..edc01f1a5b 100644 --- a/src/zarr/storage/__init__.py +++ b/src/zarr/storage/__init__.py @@ -10,6 +10,7 @@ from zarr.storage._logging import LoggingStore from zarr.storage._memory import GpuMemoryStore, ManagedMemoryStore, MemoryStore from zarr.storage._obstore import ObjectStore +from zarr.storage._url_pipeline import parse_pipeline, resolve_pipeline from zarr.storage._wrapper import WrapperStore from zarr.storage._zip import ZipStore @@ -25,6 +26,8 @@ "StorePath", "WrapperStore", "ZipStore", + "parse_pipeline", + "resolve_pipeline", ] diff --git a/src/zarr/storage/_common.py b/src/zarr/storage/_common.py index 72b5fc8a40..7bf294bfd3 100644 --- a/src/zarr/storage/_common.py +++ b/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 UPath, _join_paths, normalize_path, parse_store_url _has_fsspec = importlib.util.find_spec("fsspec") @@ -47,14 +53,22 @@ class StorePath: The store to use. path : str The path within the store. + + Attributes + ---------- + zarr_format : ZarrFormat | None + Zarr format selected by a `zarr2:`/`zarr3:` URL pipeline segment, + or None. Not part of the StorePath's identity (ignored by `__eq__`). """ store: Store path: str + zarr_format: ZarrFormat | None def __init__(self, store: Store, path: str = "") -> None: self.store = store self.path = normalize_path(path) + self.zarr_format = None @property def read_only(self) -> bool: @@ -72,7 +86,10 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No Open StorePath based on the provided mode. * If the mode is None, return an opened version of the store with no changes. - * If the mode is 'r+', 'w-', 'w', or 'a' and the store is read-only, raise a ValueError. + * If the mode is 'r+', 'w-', or 'w' and the store is read-only, raise a ValueError. + * If the mode is 'a' (open-or-create) and the store is read-only, serve the + "open" half: the StorePath is opened read-only, and any subsequent write + fails at the store level. * If the mode is 'r' and the store is not read-only, return a copy of the store with read_only set to True. * If the mode is 'w-' and the store is not read-only and the StorePath contains keys, raise a FileExistsError. * If the mode is 'w' and the store is not read-only, delete all keys nested within the StorePath. @@ -95,7 +112,7 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No FileExistsError If the mode is 'w-' and the store path already exists. ValueError - If the mode is not "r" and the store is read-only, or + If the mode is "r+", "w-", or "w" and the store is read-only, or """ # fastpath if mode is None @@ -106,8 +123,9 @@ async def open(cls, store: Store, path: str, mode: AccessModeLiteral | None = No raise ValueError(f"Invalid mode: {mode}, expected one of {ANY_ACCESS_MODE}") if store.read_only: - # Don't allow write operations on a read-only store - if mode != "r": + # mode "a" (open-or-create) on a read-only store means open-only; + # unambiguous write modes are rejected outright. + if mode not in ("r", "a"): raise ValueError( f"Store is read-only but mode is {mode!r}. Create a writable store or use 'r' mode." ) @@ -273,7 +291,9 @@ def delete_sync(self) -> None: def __truediv__(self, other: str) -> StorePath: """Combine this store path with another path""" - return self.__class__(self.store, _join_paths([self.path, other])) + result = self.__class__(self.store, _join_paths([self.path, other])) + result.zarr_format = self.zarr_format + return result def __str__(self) -> str: return _join_paths([str(self.store), self.path]) @@ -348,6 +368,16 @@ 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 normalize_path(result.path): + result.store.close() + 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 @@ -464,6 +494,13 @@ 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]) + store_path = await StorePath.open(result.store, path=combined_path, mode=mode) + store_path.zarr_format = result.zarr_format + return store_path + if isinstance(store_like, StorePath): # Already a StorePath if storage_options: diff --git a/src/zarr/storage/_url_pipeline.py b/src/zarr/storage/_url_pipeline.py new file mode 100644 index 0000000000..0161bdee6a --- /dev/null +++ b/src/zarr/storage/_url_pipeline.py @@ -0,0 +1,311 @@ +""" +Parsing and resolution of URL pipelines (https://github.com/jbms/url-pipeline). + +Importing this module is cheap and has no side effects: third-party adapters +(registered through the `zarr.url_adapters` entry-point group) are loaded +only when a pipeline URL naming their scheme is actually resolved. + +The `|` character is reserved as the pipeline delimiter in every string +store specification: a string containing `|` is always routed through the +pipeline machinery, and no percent-escape is decoded. To address a local +file whose *name* contains `|` (or `#`), pass a `pathlib.Path` instead of a +string. + +As a zarr-python extension to the specification (which requires the root +sub-URL of an absolute pipeline to carry a scheme), the root sub-URL may be +a schemeless local filesystem path, e.g. `data/example.zip|zip:`. Schemeless +roots are treated as opaque text — no query or fragment splitting is applied +to them. Such pipelines are not portable to other URL pipeline +implementations; portable pipelines should spell the root as a `file:` URL. +""" + +from __future__ import annotations + +import dataclasses +import re +from typing import TYPE_CHECKING, Any + +from zarr.abc.url_pipeline import ( + AdapterResolution, + PipelineContext, + PipelineSegment, +) +from zarr.errors import URLPipelineError +from zarr.registry import get_url_adapter, list_url_adapter_schemes +from zarr.storage._memory import ManagedMemoryStore +from zarr.storage._utils import parse_store_url + +if TYPE_CHECKING: + from zarr.abc.store import Store + from zarr.core.common import AccessModeLiteral + +__all__ = ["is_url_pipeline", "parse_pipeline", "resolve_pipeline"] + +# Adapter scheme per RFC 3986 plus "." to permit vendor-prefixed +# nonstandard schemes (e.g. "earthmover.myscheme"). +_SCHEME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9+.\-]*$") + +# Root scheme at the very start of the sub-URL. The negative lookahead +# excludes fsspec's chained-URL syntax (``zip::file://...``), which is not a +# URL pipeline and keeps flowing through the fsspec machinery. +_ROOT_SCHEME_RE = re.compile(r"([a-zA-Z][a-zA-Z0-9+.\-]*):(?!:)") + +# Schemes zarr resolves natively at the pipeline root. These are never +# dispatched to a registered root adapter, so an installed package cannot +# intercept zarr's own local-path and in-memory routing. +_NATIVE_ROOT_SCHEMES = frozenset({"file", "memory"}) + +# An absolute Windows drive path (C:\... or C:/...), which counts as an +# absolute path in a `file:` pipeline root. +_WINDOWS_DRIVE_RE = re.compile(r"[A-Za-z]:[/\\]") + + +def _root_scheme(root_sub_url: str) -> str: + """ + Detect the scheme of the root sub-URL, or `""` for an opaque root. + + Scheme extraction happens on the raw string (so nothing `urlparse` + would strip or reject — whitespace, exotic authorities — can desync the + detected scheme from the body). Single-letter candidates are delegated + to `parse_store_url`, which knows Windows drive letters are not schemes. + """ + match = _ROOT_SCHEME_RE.match(root_sub_url) + if match is None: + return "" + scheme = match.group(1) + if len(scheme) == 1: + try: + return parse_store_url(root_sub_url).scheme.lower() + except ValueError: + return scheme.lower() + return scheme.lower() + + +def _split_query(sub_url: str) -> tuple[str, str | None]: + """Split a sub-URL on the first `?`. Fragments are not supported.""" + if "#" in sub_url: + raise URLPipelineError( + f"URL pipeline sub-URLs do not support fragments: {sub_url!r}. " + "Percent-encode '#' as '%23' if it is part of the path." + ) + body, sep, query = sub_url.partition("?") + return body, query if sep else None + + +def parse_pipeline(url: str) -> tuple[PipelineSegment, ...]: + """ + Parse a URL pipeline into its `|`-delimited segments. + + The first segment is the *root* sub-URL; its scheme is detected with the + same rules as ordinary store URLs (Windows drive letters are not + schemes). Subsequent segments are *adapter* sub-URLs of the form + `scheme:body` where the trailing colon is optional when the body is + empty (`zip` is equivalent to `zip:`). + + A schemeless root (a bare local path) is accepted as a zarr-python + extension to the specification and treated as opaque text: no query or + fragment splitting applies, since `?` and `#` are ordinary filename + characters there. See the module docstring. + + Segment text is preserved verbatim (no case or percent-encoding + normalization) except that schemes are lowercased. + """ + parts = url.split("|") + if any(not part for part in parts): + raise URLPipelineError(f"URL pipeline contains an empty sub-URL: {url!r}") + + segments: list[PipelineSegment] = [] + for index, part in enumerate(parts): + if index == 0: + scheme = _root_scheme(part) + if not scheme: + # opaque root (bare local path, fsspec chained URL, ...) + segments.append(PipelineSegment(scheme="", body=part, query=None, raw=part)) + continue + body_and_scheme, query = _split_query(part) + body = body_and_scheme[len(scheme) + 1 :] + segments.append(PipelineSegment(scheme=scheme, body=body, query=query, raw=part)) + else: + body_and_scheme, query = _split_query(part) + scheme, _, body = body_and_scheme.partition(":") + scheme = scheme.lower() + if not _SCHEME_RE.match(scheme): + raise URLPipelineError( + f"invalid adapter scheme {scheme!r} in pipeline segment {part!r}" + ) + segments.append(PipelineSegment(scheme=scheme, body=body, query=query, raw=part)) + return tuple(segments) + + +def _root_routes_to_adapter(scheme: str) -> bool: + """ + Whether a root sub-URL with this scheme is dispatched to a registered + root adapter. Schemes zarr resolves natively (`file:`, `memory:`) and + opaque roots are excluded; the registry check inspects entry-point + names only — no adapter code is imported here. + """ + return ( + bool(scheme) and scheme not in _NATIVE_ROOT_SCHEMES and scheme in list_url_adapter_schemes() + ) + + +def is_url_pipeline(url: str) -> bool: + """ + Whether `url` should be routed through the URL pipeline machinery. + + True when the URL contains a `|` separator, or when its scheme has a + registered URL pipeline adapter (a *root adapter* such as `gh:`). + fsspec chained URLs (`zip::file://...`) and zarr's native `file:` / + `memory:` schemes are never routed to a root adapter. + """ + if "|" in url: + return True + return _root_routes_to_adapter(_root_scheme(url)) + + +async def resolve_pipeline( + url: str, + *, + mode: AccessModeLiteral | None = None, + storage_options: dict[str, Any] | None = None, +) -> AdapterResolution: + """ + Resolve a URL pipeline into a store and a residual path. + + Parameters + ---------- + url : str + A URL pipeline, e.g. `"s3://bucket/data.zip|zip:|zarr3:"`. + mode : AccessModeLiteral | None + The caller's access mode. `"r"` requires a read-only store; the + resolver enforces this on whatever the final adapter returns. + storage_options : dict | None + Options forwarded to the root sub-URL's store (and visible to + adapters via the context). Non-dict forms are reserved for future + per-segment configuration (one mapping per pipeline segment). + """ + segments = parse_pipeline(url) + if len(segments) == 1 and not _root_routes_to_adapter(segments[0].scheme): + raise URLPipelineError( + f"{url!r} is not a URL pipeline: it has no '|' separator and no " + f"URL pipeline adapter is registered for scheme {segments[0].scheme!r}" + ) + return await _resolve(segments, mode=mode, storage_options=storage_options) + + +async def _resolve( + segments: tuple[PipelineSegment, ...], + *, + mode: AccessModeLiteral | None, + storage_options: dict[str, Any] | None, +) -> AdapterResolution: + if len(segments) == 1 and not _root_routes_to_adapter(segments[0].scheme): + return AdapterResolution( + store=await _resolve_root(segments[0], mode=mode, storage_options=storage_options) + ) + + *preceding, last = segments + try: + adapter_cls = get_url_adapter(last.scheme) + except URLPipelineError as exc: + raise URLPipelineError( + f"{exc} Note: '|' is reserved as the URL pipeline delimiter; to " + "address a local file whose name contains '|', pass a " + "pathlib.Path instead of a string." + ) from None + + context = PipelineContext( + preceding=tuple(preceding), + mode=mode, + storage_options=storage_options, + ) + resolution = await adapter_cls.open_pipeline_segment(last, context) + if mode == "r" and not resolution.store.read_only: + # The caller required read-only; enforce it rather than trusting + # the adapter to have honored context.read_only. + try: + read_only_store = resolution.store.with_read_only(True) + except NotImplementedError as exc: + raise URLPipelineError( + f"adapter {last.scheme!r} returned a writable store for mode 'r', " + "and the store does not support read-only conversion via " + ".with_read_only()" + ) from exc + await read_only_store._ensure_open() + resolution = dataclasses.replace(resolution, store=read_only_store) + return resolution + + +async def _resolve_root( + segment: PipelineSegment, + *, + mode: AccessModeLiteral | None, + storage_options: dict[str, Any] | None, +) -> Store: + """ + Resolve the root sub-URL of a pipeline into a store. + + `memory:` and `file:` roots are resolved here with the URL pipeline + spec's semantics (spelling equivalences, mandatory absolute `file:` + paths); everything else — bare local paths, fsspec URLs — delegates to + the existing `StoreLike` machinery unchanged. + """ + if segment.scheme == "memory": + return _resolve_memory_root(segment, mode=mode, storage_options=storage_options) + from zarr.storage._common import make_store # circular import + + if segment.scheme == "file": + # Per the spec: file://localhost/p and file:///p are equivalent to + # file:/p; other authorities are unsupported; relative paths are + # forbidden; file: URLs carry no query. + if segment.query is not None: + raise URLPipelineError(f"'file:' pipeline roots do not accept a query: {segment.raw!r}") + body = segment.body + if body.startswith("//"): + authority, sep, rest = body[2:].partition("/") + if authority not in ("", "localhost"): + raise URLPipelineError( + f"unsupported authority {authority!r} in 'file:' pipeline " + f"root {segment.raw!r}; only an empty authority or " + "'localhost' is allowed" + ) + body = f"/{rest}" if sep else "" + # a file URL spells a Windows drive path as /C:/...; strip the + # leading slash so the local-path machinery sees the drive + if body.startswith("/") and _WINDOWS_DRIVE_RE.match(body[1:]): + body = body[1:] + if not (body.startswith("/") or _WINDOWS_DRIVE_RE.match(body)): + raise URLPipelineError( + f"'file:' pipeline roots must carry an absolute path: {segment.raw!r}" + ) + return await make_store(f"file:{body}", mode=mode, storage_options=storage_options) + + try: + return await make_store(segment.raw, mode=mode, storage_options=storage_options) + except ValueError as exc: + raise URLPipelineError( + f"could not resolve the pipeline root {segment.raw!r}: {exc}" + ) from exc + + +def _resolve_memory_root( + segment: PipelineSegment, + *, + mode: AccessModeLiteral | None, + storage_options: dict[str, Any] | None, +) -> Store: + """ + Resolve a `memory:` pipeline root per the spec's spelling equivalences: + `memory:` ≡ `memory:/` ≡ `memory://`, and `memory:a` ≡ `memory:/a` ≡ + `memory://a`. The first path component names the managed store; the + remainder is a path within it. + """ + if segment.query is not None: + raise URLPipelineError(f"'memory:' pipeline roots do not accept a query: {segment.raw!r}") + if storage_options: + raise TypeError( + "'storage_options' was provided but unused. " + "'storage_options' is only used when the store is passed as an FSSpec URI string.", + ) + name, _, path = segment.body.lstrip("/").partition("/") + return ManagedMemoryStore(name=name, path=path, read_only=mode == "r") diff --git a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt index 7eb0eb7c86..456c725b4e 100644 --- a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt +++ b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt @@ -13,4 +13,6 @@ another_ndbuffer = package_with_entrypoint:TestEntrypointGroup.NDBuffer [zarr.codec_pipeline] another_pipeline = package_with_entrypoint:TestEntrypointGroup.Pipeline [zarr.data_type] -new_data_type = package_with_entrypoint:TestDataType \ No newline at end of file +new_data_type = package_with_entrypoint:TestDataType +[zarr.url_adapters] +example-pkg.entrypoint-scheme = package_with_entrypoint:TestEntrypointURLAdapter diff --git a/tests/package_with_entrypoint/__init__.py b/tests/package_with_entrypoint/__init__.py index 23afcf1dc2..cad73d061d 100644 --- a/tests/package_with_entrypoint/__init__.py +++ b/tests/package_with_entrypoint/__init__.py @@ -7,6 +7,7 @@ import zarr.core.buffer from zarr.abc.codec import ArrayBytesCodec, CodecInput, CodecPipeline +from zarr.abc.url_pipeline import AdapterResolution, URLPipelineAdapter from zarr.codecs import BytesCodec from zarr.core.buffer import Buffer, NDBuffer from zarr.core.dtype.npy.bool import Bool @@ -16,6 +17,7 @@ from collections.abc import Iterable from typing import Any, ClassVar, Literal, Self + from zarr.abc.url_pipeline import PipelineContext, PipelineSegment from zarr.core.array_spec import ArraySpec from zarr.core.common import ZarrFormat from zarr.core.dtype.common import DTypeJSON, DTypeSpec_V2 @@ -100,3 +102,16 @@ def to_json(self, zarr_format: ZarrFormat) -> str | DTypeSpec_V2: # type: ignor if zarr_format == 3: return self._zarr_v3_name raise ValueError("zarr_format must be 2 or 3") + + +class TestEntrypointURLAdapter(URLPipelineAdapter): + """URL pipeline adapter discovered via the zarr.url_adapters entry point.""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + from zarr.storage import MemoryStore + + store = await MemoryStore.open(read_only=context.read_only) + return AdapterResolution(store=store, path=segment.body) diff --git a/tests/test_api.py b/tests/test_api.py index 45d0c0dee4..c4e561c180 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -198,6 +198,9 @@ async def test_open_like_creates_array_by_default( async def test_open_like_default_mode_rejects_read_only_store( zarr_format: ZarrFormat, ) -> None: + # mode "a" (open-or-create) on a read-only store serves the "open" half: + # since the node does not exist and creating is impossible, the open + # fails with a not-found error rather than rejecting the mode upfront. ref_arr = zarr.create_array( store={}, shape=(11, 12), @@ -206,7 +209,7 @@ async def test_open_like_default_mode_rejects_read_only_store( zarr_format=zarr_format, ) - with pytest.raises(ValueError, match="Store is read-only but mode is 'a'"): + with pytest.raises(ValueError, match="No array found"): await zarr.api.asynchronous.open_like( ref_arr, path="foo", diff --git a/tests/test_url_pipeline/__init__.py b/tests/test_url_pipeline/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/test_url_pipeline/conftest.py b/tests/test_url_pipeline/conftest.py new file mode 100644 index 0000000000..13c717e04e --- /dev/null +++ b/tests/test_url_pipeline/conftest.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +import zarr.registry + +if TYPE_CHECKING: + from collections.abc import Generator + + +@pytest.fixture +def clean_url_adapter_registry() -> Generator[None, None, None]: + """Snapshot and restore the URL adapter registry around a test.""" + registry = zarr.registry._url_adapter_registry + saved = dict(registry) + saved_lazy = list(registry.lazy_load_list) + yield + registry.clear() + registry.update(saved) + registry.lazy_load_list[:] = saved_lazy diff --git a/tests/test_url_pipeline/test_parser.py b/tests/test_url_pipeline/test_parser.py new file mode 100644 index 0000000000..13d7846e85 --- /dev/null +++ b/tests/test_url_pipeline/test_parser.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import sys + +import pytest +from hypothesis import given +from hypothesis import strategies as st + +from zarr.errors import URLPipelineError +from zarr.storage._url_pipeline import parse_pipeline + + +def test_single_root_url() -> None: + (segment,) = parse_pipeline("s3://bucket/key") + assert segment.scheme == "s3" + assert segment.body == "//bucket/key" + assert segment.query is None + assert segment.raw == "s3://bucket/key" + assert str(segment) == "s3://bucket/key" + + +def test_schemeless_root() -> None: + (segment,) = parse_pipeline("/local/path") + assert segment.scheme == "" + assert segment.body == "/local/path" + assert segment.raw == "/local/path" + + +def test_adapter_chain() -> None: + segments = parse_pipeline("s3://bucket/data.zip|zip:inner/path|zarr3:") + assert [s.scheme for s in segments] == ["s3", "zip", "zarr3"] + assert segments[1].body == "inner/path" + assert segments[2].body == "" + + +def test_trailing_colon_optional() -> None: + with_colon = parse_pipeline("file:/tmp/x.zip|zip:") + without_colon = parse_pipeline("file:/tmp/x.zip|zip") + assert with_colon[1].scheme == without_colon[1].scheme == "zip" + assert with_colon[1].body == without_colon[1].body == "" + + +def test_scheme_case_insensitive() -> None: + segments = parse_pipeline("FILE:/tmp/x.zip|ZIP:Inner/Path") + assert segments[0].scheme == "file" + assert segments[1].scheme == "zip" + # bodies are case-preserved + assert segments[1].body == "Inner/Path" + + +def test_case_preserved_in_raw() -> None: + # e.g. icechunk snapshot IDs are case-significant + segments = parse_pipeline("file:/tmp/repo|icechunk://ABCDEFGH12345678ABCD/x") + assert segments[1].raw == "icechunk://ABCDEFGH12345678ABCD/x" + assert segments[1].body == "//ABCDEFGH12345678ABCD/x" + + +def test_vendor_prefixed_scheme() -> None: + segments = parse_pipeline("file:/data|vendor-1.custom+adapter:sub/path") + assert segments[1].scheme == "vendor-1.custom+adapter" + assert segments[1].body == "sub/path" + + +def test_query_strings() -> None: + segments = parse_pipeline("https://example.com/d.zip?token=abc|zip:x?opt=1") + assert segments[0].query == "token=abc" + assert segments[0].body == "//example.com/d.zip" + assert segments[1].query == "opt=1" + assert segments[1].body == "x" + + +def test_empty_query() -> None: + (segment,) = parse_pipeline("https://example.com/d?") + assert segment.query == "" + + +def test_windows_drive_path_is_not_a_scheme() -> None: + # On Windows parse_store_url treats C:\... as a local path; elsewhere the + # single-letter scheme is preserved but must not crash the parser. + (segment,) = parse_pipeline(r"C:\data\store") + assert segment.raw == r"C:\data\store" + if sys.platform == "win32": + assert segment.scheme == "" + assert segment.body == r"C:\data\store" + else: + assert segment.scheme == "c" + + +@pytest.mark.parametrize("url", ["a||b:", "|zip:", "file:/tmp|", ""]) +def test_empty_sub_url_rejected(url: str) -> None: + with pytest.raises(URLPipelineError, match="empty sub-URL"): + parse_pipeline(url) + + +def test_fragment_rejected() -> None: + with pytest.raises(URLPipelineError, match="fragment"): + parse_pipeline("file:/tmp/x.zip|zip:inner#frag") + + +@pytest.mark.parametrize("segment", ["1zip:", "zi p:x", "zip@:x"]) +def test_invalid_adapter_scheme_rejected(segment: str) -> None: + with pytest.raises(URLPipelineError, match="invalid adapter scheme"): + parse_pipeline(f"file:/tmp/x|{segment}") + + +def test_single_letter_scheme_with_exotic_authority() -> None: + # single-letter candidates are delegated to parse_store_url, which may + # reject an exotic authority; the raw scheme is used as the fallback + segments = parse_pipeline("x://[authority]/path|zip:") + assert segments[0].scheme in ("", "x") + assert segments[0].raw == "x://[authority]/path" + + +def test_fsspec_chained_root_is_opaque() -> None: + # fsspec's ``scheme::`` chaining is not pipeline syntax; such roots are + # schemeless/opaque and flow through the ordinary store machinery + (segment,) = parse_pipeline("zip::file:///tmp/data.zip") + assert segment.scheme == "" + assert segment.body == "zip::file:///tmp/data.zip" + + +def test_whitespace_prefixed_root_is_opaque() -> None: + # urlparse strips leading whitespace when sniffing a scheme; the parser + # must not, or the detected scheme desyncs from the body slice + segments = parse_pipeline("\tfile:/tmp/x|zip:") + assert segments[0].scheme == "" + assert segments[0].body == "\tfile:/tmp/x" + + +def test_schemeless_root_keeps_query_and_fragment_chars() -> None: + # ? and # are ordinary filename characters in a bare local path + segments = parse_pipeline("/tmp/d?v=1|zip:") + assert segments[0].scheme == "" + assert segments[0].body == "/tmp/d?v=1" + assert segments[0].query is None + segments = parse_pipeline("/tmp/d#frag|zip:") + assert segments[0].body == "/tmp/d#frag" + + +def test_round_trip() -> None: + url = "s3://bucket/a.zip?v=2|zip:b/inner.zip|zip:c|zarr3:" + segments = parse_pipeline(url) + assert "|".join(s.raw for s in segments) == url + + +# --- property-based tests ----------------------------------------------- +# +# These exercise the parser's *splitting invariants*, not URL pipeline spec +# validity: parse_pipeline is a permissive segment splitter (bodies and +# queries are opaque text to it), and semantic validation is left to the +# resolver and the adapters. Rather than enumerating registered schemes, +# the strategies sample the full scheme grammar the parser accepts +# (RFC 3986 plus "." for vendor prefixes), so every valid root/adapter +# scheme is reachable. Bodies and queries draw from printable ASCII minus +# the characters the parser itself splits on ("|", "#", and "?" for +# bodies); the generated text is not required to be a well-formed URI. + +_ADAPTER_SCHEME = st.from_regex(r"[a-zA-Z][a-zA-Z0-9+.\-]{0,15}", fullmatch=True) +# two or more characters, so Windows drive-letter handling cannot reclassify +# the root scheme as a local path on one platform but not another +_ROOT_SCHEME = st.from_regex(r"[a-zA-Z]{2}[a-zA-Z0-9+.\-]{0,14}", fullmatch=True) +_BODY = st.text(st.characters(min_codepoint=32, max_codepoint=126, exclude_characters="|#?")) +# a root body starting with ":" would spell fsspec's chained-URL syntax +# ("scheme::..."), which the parser deliberately treats as an opaque root +_ROOT_BODY = _BODY.filter(lambda body: not body.startswith(":")) +_QUERY = st.text(st.characters(min_codepoint=32, max_codepoint=126, exclude_characters="|#")) + + +@st.composite +def pipelines( + draw: st.DrawFn, min_depth: int = 1 +) -> tuple[str, list[tuple[str, str, str | None, str]]]: + """A parseable pipeline URL up to depth 8, with its expected split.""" + depth = draw(st.integers(min_value=min_depth, max_value=8)) + expected = [] + parts = [] + for index in range(depth): + scheme = draw(_ROOT_SCHEME if index == 0 else _ADAPTER_SCHEME) + body = draw(_ROOT_BODY if index == 0 else _BODY) + query = draw(st.none() | _QUERY) + raw = f"{scheme}:{body}" + (f"?{query}" if query is not None else "") + parts.append(raw) + expected.append((scheme.lower(), body, query, raw)) + return "|".join(parts), expected + + +@given(pipelines()) +def test_valid_pipeline_invariants( + case: tuple[str, list[tuple[str, str, str | None, str]]], +) -> None: + url, expected = case + segments = parse_pipeline(url) + # schemes lowercased; bodies and queries preserved verbatim + assert [(s.scheme, s.body, s.query, s.raw) for s in segments] == expected + # lossless round trip + assert "|".join(s.raw for s in segments) == url + + +@given(pipelines(), st.data()) +def test_property_empty_sub_url_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + parts.insert(data.draw(st.integers(0, len(parts))), "") + with pytest.raises(URLPipelineError, match="empty sub-URL"): + parse_pipeline("|".join(parts)) + + +@given(pipelines(min_depth=2), st.data()) +def test_property_invalid_adapter_scheme_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + # corrupt one adapter segment's scheme with a leading character the + # scheme grammar forbids + index = data.draw(st.integers(1, len(parts) - 1)) + parts[index] = data.draw(st.sampled_from(["1", " ", "@", "~"])) + parts[index] + with pytest.raises(URLPipelineError, match="invalid adapter scheme"): + parse_pipeline("|".join(parts)) + + +@given(pipelines(), st.data()) +def test_property_fragment_rejected( + case: tuple[str, list[tuple[str, str, str | None, str]]], data: st.DataObject +) -> None: + url, _ = case + parts = url.split("|") + parts[data.draw(st.integers(0, len(parts) - 1))] += "#frag" + with pytest.raises(URLPipelineError, match="fragment"): + parse_pipeline("|".join(parts)) diff --git a/tests/test_url_pipeline/test_resolver.py b/tests/test_url_pipeline/test_resolver.py new file mode 100644 index 0000000000..198c178764 --- /dev/null +++ b/tests/test_url_pipeline/test_resolver.py @@ -0,0 +1,569 @@ +from __future__ import annotations + +import dataclasses +import sys +from typing import TYPE_CHECKING, ClassVar + +import pytest + +import zarr +import zarr.registry +from zarr.abc.store import Store +from zarr.abc.url_pipeline import ( + AdapterResolution, + PipelineContext, + PipelineSegment, + URLPipelineAdapter, +) +from zarr.errors import URLPipelineError, ZarrUserWarning +from zarr.registry import ( + get_url_adapter, + list_url_adapter_schemes, + register_url_adapter, +) +from zarr.storage import ManagedMemoryStore, MemoryStore, WrapperStore +from zarr.storage._common import make_store, make_store_path +from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline +from zarr.storage._utils import _join_paths + +if TYPE_CHECKING: + from pathlib import Path + +pytestmark = pytest.mark.usefixtures("clean_url_adapter_registry") + + +class TracingStore(WrapperStore[Store]): + """Wrapper that records the context it was created from.""" + + context: PipelineContext + segment: PipelineSegment + + +class WrapperAdapter(URLPipelineAdapter): + """ + A wrapper-style adapter: resolves the preceding pipeline into a store. + + Follows the wrapper contract: the preceding resolution's residual path + is carried forward (joined with this segment's own path), and unchanged + fields survive via `dataclasses.replace`. + """ + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + preceding = await context.resolve_preceding() + store = TracingStore(preceding.store) + store.context = context + store.segment = segment + return dataclasses.replace( + preceding, store=store, path=_join_paths([preceding.path, segment.body]) + ) + + +class NativeAdapter(URLPipelineAdapter): + """A native-style adapter: consumes the preceding URL as a string.""" + + seen_urls: ClassVar[list[str]] = [] + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + cls.seen_urls.append(context.preceding_url) + store = await MemoryStore.open(read_only=context.read_only) + return AdapterResolution(store=store, path=segment.body) + + +class RootAdapter(URLPipelineAdapter): + """A root-scheme adapter (no preceding segments), like al://.""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + assert context.preceding == () + if context.mode in ("w", "w-", "r+"): + raise ValueError("read-only scheme") + store = await MemoryStore.open(read_only=True) + return AdapterResolution(store=store, path=segment.body.lstrip("/")) + + +class DisobedientAdapter(URLPipelineAdapter): + """An adapter that ignores `context.read_only` (a contract violation).""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + return AdapterResolution(store=await MemoryStore.open(read_only=False)) + + +class FormatAdapter(URLPipelineAdapter): + """A dummy format-selecting adapter, like the builtin `zarr2:`.""" + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + preceding = await context.resolve_preceding() + return dataclasses.replace(preceding, zarr_format=2) + + +class TestRegistry: + def test_register_and_get(self) -> None: + register_url_adapter("demo", WrapperAdapter) + assert get_url_adapter("demo") is WrapperAdapter + assert get_url_adapter("DEMO") is WrapperAdapter + assert "demo" in list_url_adapter_schemes() + + def test_unknown_scheme(self) -> None: + with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"): + get_url_adapter("nonexistent-scheme") + + def test_reregistering_scheme_warns(self) -> None: + register_url_adapter("demo", WrapperAdapter) + with pytest.warns(ZarrUserWarning, match="is being replaced"): + register_url_adapter("demo", NativeAdapter) + assert get_url_adapter("demo") is NativeAdapter + # re-registering the same class is not a collision + register_url_adapter("demo", NativeAdapter) + + @pytest.mark.usefixtures("set_path") + def test_entrypoint_discovery(self) -> None: + assert "example-pkg.entrypoint-scheme" in list_url_adapter_schemes() + cls = get_url_adapter("example-pkg.entrypoint-scheme") + assert cls.__name__ == "TestEntrypointURLAdapter" + + @pytest.mark.usefixtures("set_path") + async def test_entrypoint_end_to_end(self) -> None: + result = await resolve_pipeline("memory://src|example-pkg.entrypoint-scheme:sub/path") + assert result.path == "sub/path" + + @pytest.mark.usefixtures("set_path") + def test_entrypoint_scheme_lookup_is_case_insensitive(self) -> None: + # entry-point names are matched case-insensitively, like schemes + cls = get_url_adapter("Example-PKG.Entrypoint-Scheme") + assert cls.__name__ == "TestEntrypointURLAdapter" + + @pytest.mark.usefixtures("set_path") + def test_loading_one_scheme_leaves_others_pending(self) -> None: + # resolving one scheme must not import other providers' entry points + registry = zarr.registry._url_adapter_registry + assert any(e.name == "example-pkg.entrypoint-scheme" for e in registry.lazy_load_list) + with pytest.raises(URLPipelineError, match="no URL pipeline adapter"): + get_url_adapter("some-other-scheme") + assert any(e.name == "example-pkg.entrypoint-scheme" for e in registry.lazy_load_list) + assert get_url_adapter("example-pkg.entrypoint-scheme").__name__ == ( + "TestEntrypointURLAdapter" + ) + + +class TestIsURLPipeline: + def test_pipe_routes(self) -> None: + assert is_url_pipeline("memory://x|demo:") + + def test_registered_root_scheme_routes(self) -> None: + register_url_adapter("rooty", RootAdapter) + assert is_url_pipeline("rooty://org/repo") + + @pytest.mark.parametrize("url", ["s3://bucket/key", "/local/path", "memory://x", "C:.zarr"]) + def test_plain_urls_do_not_route(self, url: str) -> None: + assert not is_url_pipeline(url) + + def test_fsspec_chained_urls_do_not_route(self) -> None: + # fsspec's ``scheme::`` chaining is not a URL pipeline: even with a + # same-named root adapter registered, these keep flowing to fsspec. + register_url_adapter("zip", RootAdapter) + assert not is_url_pipeline("zip::file:///tmp/data.zip") + assert not is_url_pipeline("simplecache::s3://bucket/key") + + def test_native_schemes_do_not_route_to_adapters(self) -> None: + # an installed package must not be able to intercept zarr's own + # local-path and in-memory routing by registering file:/memory: + register_url_adapter("file", RootAdapter) + register_url_adapter("memory", RootAdapter) + assert not is_url_pipeline("file:/tmp/x") + assert not is_url_pipeline("memory://x") + + def test_exotic_authority_root_scheme(self) -> None: + # The spec's own root-URL example: urlparse rejects the bracketed + # non-IP authority, so scheme detection must work on the raw string + # rather than raising before routing. + url = "vendor1-2.custom-1+proto.ext://[authority]/path?query/part?x" + assert not is_url_pipeline(url) + register_url_adapter("vendor1-2.custom-1+proto.ext", RootAdapter) + assert is_url_pipeline(url) + + +class TestResolve: + async def test_wrapper_adapter_chain(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{tmp_path}|wrap:inner/path") + assert isinstance(result.store, TracingStore) + assert result.path == "inner/path" + assert result.store.context.preceding_url == str(tmp_path) + + async def test_nested_wrappers_preserve_residual_paths(self, tmp_path: Path) -> None: + # each wrapper joins the preceding residual path with its own, so + # no segment's path is lost in root|wrap:a|wrap:b + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{tmp_path}|wrap:a|wrap:b") + assert result.path == "a/b" + + async def test_native_adapter_gets_preceding_url(self) -> None: + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await resolve_pipeline("s3://bucket/repo|native:") + assert NativeAdapter.seen_urls == ["s3://bucket/repo"] + + async def test_multi_segment_preceding_url(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await resolve_pipeline("memory://base|wrap:a|native:x") + assert NativeAdapter.seen_urls == ["memory://base|wrap:a"] + + async def test_root_adapter(self) -> None: + register_url_adapter("rooty", RootAdapter) + result = await resolve_pipeline("rooty://org/repo") + assert result.path == "org/repo" + assert result.store.read_only + + async def test_root_adapter_composes_with_chain(self) -> None: + register_url_adapter("rooty", RootAdapter) + register_url_adapter("native", NativeAdapter) + NativeAdapter.seen_urls.clear() + await resolve_pipeline("rooty://org/repo|native:x") + assert NativeAdapter.seen_urls == ["rooty://org/repo"] + + async def test_read_only_flag(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{tmp_path}|wrap:", mode="r") + assert isinstance(result.store, TracingStore) + assert result.store.context.read_only + assert result.store.context.mode == "r" + result = await resolve_pipeline(f"{tmp_path}|wrap:") + assert isinstance(result.store, TracingStore) + assert not result.store.context.read_only + assert result.store.context.mode is None + + async def test_read_only_is_enforced_on_disobedient_adapters(self) -> None: + # the resolver downgrades a writable store returned under mode "r" + # instead of trusting the adapter to have honored context.read_only + register_url_adapter("bad", DisobedientAdapter) + result = await resolve_pipeline("memory://base|bad:", mode="r") + assert result.store.read_only + store = await make_store("memory://base|bad:", mode="r") + assert store.read_only + + async def test_read_only_enforcement_without_conversion_raises(self) -> None: + # a disobedient adapter whose store cannot be converted read-only + class StubbornStore(MemoryStore): + def with_read_only(self, read_only: bool = False) -> MemoryStore: + raise NotImplementedError + + class StubbornAdapter(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + return AdapterResolution(store=await StubbornStore.open(read_only=False)) + + register_url_adapter("stubborn", StubbornAdapter) + with pytest.raises(URLPipelineError, match="does not support read-only conversion"): + await resolve_pipeline("memory://base|stubborn:", mode="r") + + async def test_resolve_preceding_mode_override(self, tmp_path: Path) -> None: + # a wrapper that only reads the preceding resource can open it + # read-only regardless of the caller's mode + class ReadOnlyRootWrapper(WrapperAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + preceding = await context.resolve_preceding(mode="r") + assert preceding.store.read_only + return dataclasses.replace(preceding, path=segment.body) + + register_url_adapter("rowrap", ReadOnlyRootWrapper) + result = await resolve_pipeline(f"{tmp_path}|rowrap:x", mode="w") + assert result.path == "x" + + async def test_resolve_preceding_storage_options_override(self, tmp_path: Path) -> None: + # an adapter that consumed its namespaced keys strips them before + # the root store ever sees them + class ConsumingWrapper(WrapperAdapter): + seen: ClassVar[object | None] = None + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + options = dict(context.storage_options or {}) + cls.seen = options.pop("consuming_secret", None) + preceding = await context.resolve_preceding(storage_options=options or None) + return dataclasses.replace(preceding, path=segment.body) + + register_url_adapter("consuming", ConsumingWrapper) + # the local-path root rejects any surviving storage_options, so this + # passing proves the adapter's keys were stripped before resolution + result = await resolve_pipeline( + f"{tmp_path}|consuming:x", storage_options={"consuming_secret": "s3cr3t"} + ) + assert result.path == "x" + assert ConsumingWrapper.seen == "s3cr3t" + + async def test_storage_options_visible_to_adapter(self) -> None: + register_url_adapter("native", NativeAdapter) + + class OptionsProbe(NativeAdapter): + seen_options: dict[str, object] | None = None + + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + cls.seen_options = context.storage_options + return await super().open_pipeline_segment(segment, context) + + register_url_adapter("probe", OptionsProbe) + opts = {"anon": True} + await resolve_pipeline("s3://bucket/x|probe:", storage_options=opts) + assert OptionsProbe.seen_options == opts + + async def test_wrapper_at_root_position_raises(self) -> None: + # a wrapper adapter used as the pipeline root has nothing to wrap + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="no preceding sub-URL"): + await resolve_pipeline("wrap:whatever") + + async def test_not_a_pipeline_raises(self) -> None: + with pytest.raises(URLPipelineError, match="is not a URL pipeline"): + await resolve_pipeline("s3://bucket/plain") + + async def test_unknown_adapter_scheme_raises(self) -> None: + with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"): + await resolve_pipeline("memory://base|no-such-adapter:") + + async def test_unknown_adapter_error_hints_at_reserved_pipe(self, tmp_path: Path) -> None: + # a filename containing '|' routes here; the error points at the fix + with pytest.raises(URLPipelineError, match="pass a\\s+pathlib.Path"): + await resolve_pipeline(f"{tmp_path}/a|b") + + async def test_base_case_value_error_is_wrapped(self) -> None: + # a root that only the fallback scheme detection accepts must not + # leak urlparse's ValueError out of the base case + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="could not resolve the pipeline root"): + await resolve_pipeline("vendor.x://[authority]/path|wrap:") + + async def test_zarr_format_flows_from_adapter(self, tmp_path: Path) -> None: + register_url_adapter("fmt2", FormatAdapter) + result = await resolve_pipeline(f"{tmp_path}|fmt2:") + assert result.zarr_format == 2 + # wrapper adapters carry the format through (dataclasses.replace) + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{tmp_path}|fmt2:|wrap:") + assert result.zarr_format == 2 + + +class TestSpecRoots: + """The spec's `memory:` / `file:` root semantics inside pipelines.""" + + @pytest.mark.parametrize("root", ["memory:", "memory:/", "memory://"]) + async def test_memory_root_spellings_equivalent(self, root: str) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{root}|wrap:") + assert isinstance(result.store, TracingStore) + inner = result.store._store + assert isinstance(inner, ManagedMemoryStore) + assert inner._name == "" + assert inner.path == "" + + @pytest.mark.parametrize("root", ["memory:a/b", "memory:/a/b", "memory://a/b"]) + async def test_memory_root_named_spellings_equivalent(self, root: str) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline(f"{root}|wrap:") + assert isinstance(result.store, TracingStore) + inner = result.store._store + assert isinstance(inner, ManagedMemoryStore) + assert inner._name == "a" + assert inner.path == "b" + + async def test_memory_root_shares_data_across_spellings(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + group = zarr.open_group("memory:pipe-share|wrap:", mode="w") + group.create_array("x", shape=(2,), dtype="i4") + reopened = zarr.open_group("memory://pipe-share|wrap:", mode="r") + assert "x" in reopened + + async def test_memory_root_rejects_query(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="do not accept a query"): + await resolve_pipeline("memory:a?opt=1|wrap:") + + async def test_memory_root_rejects_storage_options(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(TypeError, match="'storage_options' was provided but unused"): + await resolve_pipeline("memory:a|wrap:", storage_options={"anon": True}) + + async def test_file_root_localhost_authority(self, tmp_path: Path) -> None: + # file://localhost/p and file:///p are equivalent to file:/p + register_url_adapter("wrap", WrapperAdapter) + # spell the path URL-style: on Windows C:/... gains a leading slash + posix = tmp_path.as_posix() + url_path = posix if posix.startswith("/") else f"/{posix}" + zarr.open_group(f"file://localhost{url_path}|wrap:", mode="w") + opened = zarr.open_group(f"file:{tmp_path}|wrap:", mode="r") + assert isinstance(opened, zarr.Group) + zarr.open_group(f"file://{url_path}|wrap:", mode="r") + + async def test_file_root_windows_drive_spellings( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Windows drive paths count as absolute, and the URL spelling + # file:///C:/... has its leading slash stripped before the drive + register_url_adapter("wrap", WrapperAdapter) + if sys.platform == "win32": + drive_path = tmp_path.as_posix() # C:/Users/... + else: + # off Windows a drive path is just an odd relative directory; + # chdir into tmp so it is created there + monkeypatch.chdir(tmp_path) + drive_path = "C:/drive/data" + r1 = await resolve_pipeline(f"file:{drive_path}|wrap:") + r2 = await resolve_pipeline(f"file:///{drive_path}|wrap:") + assert isinstance(r1.store, TracingStore) + assert isinstance(r2.store, TracingStore) + + async def test_file_root_rejects_other_authority(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="unsupported authority"): + await resolve_pipeline("file://example.com/tmp/x|wrap:") + + @pytest.mark.parametrize("root", ["file:relative/path", "file://localhost"]) + async def test_file_root_rejects_relative_paths(self, root: str) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="absolute path"): + await resolve_pipeline(f"{root}|wrap:") + + async def test_file_root_rejects_query(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="do not accept a query"): + await resolve_pipeline(f"file:{tmp_path}?v=1|wrap:") + + +class TestMakeStoreIntegration: + async def test_make_store_path_combines_paths(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + store_path = await make_store_path(f"{tmp_path}|wrap:residual", path="user/sub") + assert store_path.path == "residual/user/sub" + + async def test_make_store_rejects_residual_path(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="resolves to a path inside a store"): + await make_store(f"{tmp_path}|wrap:residual") + + async def test_make_store_rejects_slash_root_path(self, tmp_path: Path) -> None: + # an adapter returning path="/" addresses the store root; the raw + # value is normalized before the residual-path rejection + class SlashRootAdapter(WrapperAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + preceding = await context.resolve_preceding() + return dataclasses.replace(preceding, path="/") + + register_url_adapter("slashy", SlashRootAdapter) + store = await make_store(f"{tmp_path}|slashy:") + assert store is not None + + async def test_make_store_closes_store_on_residual_path_error(self) -> None: + closed: list[bool] = [] + + class ClosingStore(MemoryStore): + def close(self) -> None: + closed.append(True) + super().close() + + class LeakProbe(URLPipelineAdapter): + @classmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + return AdapterResolution(store=await ClosingStore.open(), path="residual") + + register_url_adapter("leaky", LeakProbe) + with pytest.raises(URLPipelineError, match="resolves to a path inside a store"): + await make_store("memory://base|leaky:") + assert closed == [True] + + async def test_make_store_no_residual_path(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + store = await make_store(f"{tmp_path}|wrap:") + assert isinstance(store, TracingStore) + + async def test_zarr_open_end_to_end(self, tmp_path: Path) -> None: + register_url_adapter("wrap", WrapperAdapter) + group = zarr.open_group(f"{tmp_path}|wrap:", mode="w") + array = group.create_array("x", shape=(4,), dtype="i4") + array[:] = [1, 2, 3, 4] + assert zarr.open_array(f"{tmp_path}|wrap:x")[2] == 3 + + async def test_pipeline_store_read_only_mode(self) -> None: + register_url_adapter("native", NativeAdapter) + store_path = await make_store_path("memory://base|native:", mode="r") + assert store_path.read_only + + async def test_mode_a_downgrades_to_read_only_open(self) -> None: + # mode "a" (the zarr.open default) is open-or-create: a pipeline + # that resolves to a read-only store serves the "open" half instead + # of failing outright. + register_url_adapter("rooty", RootAdapter) + store_path = await make_store_path("rooty://org/repo", mode="a") + assert store_path.read_only + + async def test_mode_a_downgrade_applies_to_plain_stores_too(self) -> None: + # the open-or-create downgrade lives in StorePath.open, so it is + # uniform across pipeline and non-pipeline stores + store = await MemoryStore.open(read_only=True) + store_path = await make_store_path(store, mode="a") + assert store_path.read_only + + async def test_explicit_write_mode_reaches_adapter(self) -> None: + register_url_adapter("rooty", RootAdapter) + with pytest.raises(ValueError, match="read-only scheme"): + await make_store_path("rooty://org/repo", mode="w") + + async def test_storage_options_forwarded_to_root(self) -> None: + # storage_options reach the root sub-URL via resolve_preceding -> + # make_store. A local-path root does not accept storage_options, so + # forwarding them must raise the same TypeError as a non-pipeline open. + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(TypeError, match="'storage_options' was provided but unused"): + await make_store("/tmp/some/path|wrap:", storage_options={"anon": True}) + + async def test_store_path_truediv_keeps_format(self, tmp_path: Path) -> None: + # StorePath.__truediv__ constructs a new instance; the pipeline- + # selected format must survive the division + register_url_adapter("fmt2", FormatAdapter) + store_path = await make_store_path(f"{tmp_path}|fmt2:") + assert store_path.zarr_format == 2 + assert (store_path / "sub/node").zarr_format == 2 + + async def test_zarr_format_merges_into_open(self, tmp_path: Path) -> None: + register_url_adapter("fmt2", FormatAdapter) + group = zarr.open_group(f"{tmp_path}|fmt2:", mode="w") + assert group.metadata.zarr_format == 2 + + async def test_conflicting_explicit_format_raises(self, tmp_path: Path) -> None: + register_url_adapter("fmt2", FormatAdapter) + with pytest.raises(ValueError, match="conflicts with"): + zarr.open_group(f"{tmp_path}|fmt2:", mode="w", zarr_format=3) + + async def test_matching_explicit_format_ok(self, tmp_path: Path) -> None: + register_url_adapter("fmt2", FormatAdapter) + group = zarr.open_group(f"{tmp_path}|fmt2:", mode="w", zarr_format=2) + assert group.metadata.zarr_format == 2 diff --git a/tests/test_url_pipeline/test_spec_examples.py b/tests/test_url_pipeline/test_spec_examples.py new file mode 100644 index 0000000000..5538074bea --- /dev/null +++ b/tests/test_url_pipeline/test_spec_examples.py @@ -0,0 +1,258 @@ +""" +Conformance tests against the URL pipeline specification. + +Every entry in `SPEC_EXAMPLES` is an example URL from the specification +repository (https://github.com/jbms/url-pipeline @ 1a01ce6), extracted +with the same rule as the spec's own linter (`scripts/lint.py`): backticked +examples on bullet lines, skipping spans without a `:` or `|`. The spec +validates each example (and an uppercased-scheme variant) against its ABNF +grammar, so this corpus is grammar-valid by construction. + +The parser must accept every example — including schemes zarr-python has no +adapter for, since parsing is independent of adapter availability — split it +into the expected sub-URL schemes, and preserve the text losslessly. + +To regenerate after a spec update: extract examples per the rule above and +recompute the expected scheme tuples with `parse_pipeline`, reviewing any +changes against the spec diff. +""" + +from __future__ import annotations + +import re + +import pytest + +from zarr.storage._url_pipeline import parse_pipeline + +SPEC_EXAMPLES: list[tuple[str, tuple[str, ...]]] = [ + # README.md + ("s3://bucket/path/to/archive.zip|zip:path/within/zip.zarr/|zarr3:", ("s3", "zip", "zarr3")), + ( + "file:///tmp/dataset.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database", + ("file", "ocdbt"), + ), + ( + "s3+https://example.com/path/to/database.icechunk/|icechunk://tag.v5/path/to/node/|zarr3:", + ("s3+https", "icechunk", "zarr3"), + ), + ( + "vendor1-2.custom-1+proto.ext://[authority]/path?query/part?x", + ("vendor1-2.custom-1+proto.ext",), + ), + ( + "http://example.com/file|vendor1-2.custom+adapter:/path/within/adapter", + ("http", "vendor1-2.custom+adapter"), + ), + ("a.b:?", ("a.b",)), + ("http://somehost/downloads/somefile.zip|zip:", ("http", "zip")), + ("http://example.com/archive.jar|zip:path/to/file.txt", ("http", "zip")), + ("https://host/archive.zip|zip:path/in/outer.zip|zip:path/in/inner", ("https", "zip", "zip")), + ("file:///path/to/archive.zip|zip:path/within/archive", ("file", "zip")), + # avif.md + ("file:/path/to/image.avif|avif:", ("file", "avif")), + ("file:/path/to/image.avif|avif", ("file", "avif")), + # bmp.md + ("file:/path/to/image.bmp|bmp:", ("file", "bmp")), + ("file:/path/to/image.bmp|bmp", ("file", "bmp")), + # byte-range.md + ("file:/path/to/data|byte-range:1000-2000", ("file", "byte-range")), + ("file:/path/to/data|byte-range:0-1", ("file", "byte-range")), + # file.md + ("file:/", ("file",)), + ("file:/a", ("file",)), + ("file:/path/to/file.txt", ("file",)), + ("file://localhost/path/to/file.txt", ("file",)), + ("file://LOCALHOST/path/to/file.txt", ("file",)), + ("file:///path/to/file.txt", ("file",)), + ("file://somehost/sharename/path/to/file.txt", ("file",)), + # gs.md + ("gs://bucket", ("gs",)), + ("gs://bucket/", ("gs",)), + ("gs://bucket/path/within/bucket", ("gs",)), + ("gs://aaa", ("gs",)), + ( + "gs://label1-is-sixty-two-characters-long-xxxxxxxxxxxxxxxxxxxxxxxxxx.label2-is-sixty-two-characters-long-yyyyyyyyyyyyyyyyyyyyyyyyyy.label3-is-sixty-two-characters-long-zzzzzzzzzzzzzzzzzzzzzzzzzz.label4-is-thirty-one-characters", + ("gs",), + ), + # gzip.md + ("file:/path/to/data.gz|gzip:", ("file", "gzip")), + ("file:/path/to/data.gz|gzip", ("file", "gzip")), + # hdf5.md + ("s3://bucket/path/to/file.h5|hdf5:/path/to/dataset", ("s3", "hdf5")), + ("s3://bucket/path/to/file.h5|hdf5:a", ("s3", "hdf5")), + ("file:///path/to/file.h5|hdf5:", ("file", "hdf5")), + ("file:///path/to/file.h5|hdf5", ("file", "hdf5")), + # http.md + ("https://example.com/path/to/resource%20name", ("https",)), + ("https://example.com/path/to/resource?query=value", ("https",)), + ("https://example.com/path/to/?query=value", ("https",)), + ("https://example.com/path/with:colon", ("https",)), + ("https://server.example.com:1234/path/to/array", ("https",)), + ("http://a", ("http",)), + ("http://example.com", ("http",)), + ("http://local%68ost", ("http",)), + ("http://example.com:", ("http",)), + ("http://192.168.10.1:1234", ("http",)), + ("http://[::1]", ("http",)), + ("http://[::ffff:127.0.0.1]", ("http",)), + ("http://example.com/", ("http",)), + # icechunk.md + ("file:///path/to/repo.zarr.icechunk/|icechunk:", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.main/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:a", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:/path/to/node/", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/zarr.json", ("file", "icechunk")), + ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/c/0/0/1", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://branch.mybranch/path/to/node/", + ("file", "icechunk"), + ), + ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.a", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://tag.mytag/path/to/node/", + ("file", "icechunk"), + ), + ("file:///path/to/repo.zarr.icechunk/|icechunk://tag.a", ("file", "icechunk")), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://FWWFQGAW742XMX0F5MF0/path/to/node/", + ("file", "icechunk"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://ABCDEFGHJKMNPQRSTVWX/path/to/node/", + ("file", "icechunk"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk:|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://branch.other/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://tag.v5/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + ( + "file:///path/to/repo.zarr.icechunk/|icechunk://4N0217AZA4VNPYD0HR0G/|zarr3:path/to/array/", + ("file", "icechunk", "zarr3"), + ), + # jpeg.md + ("file:/path/to/image.jpeg|jpeg:", ("file", "jpeg")), + ("file:/path/to/image.jpeg|jpeg", ("file", "jpeg")), + # json.md + ("file:/path/to/data.json|json:", ("file", "json")), + ("file:/path/to/data.json|json", ("file", "json")), + ("file:/path/to/data.json|json:/path/to/node", ("file", "json")), + ("file:/path/to/data.json|json:/abc~0def", ("file", "json")), + ("file:/path/to/data.json|json:/abc~1def", ("file", "json")), + ("file:/path/to/data.json|json:/", ("file", "json")), + # memory.md + ("memory:", ("memory",)), + ("memory:/", ("memory",)), + ("memory://", ("memory",)), + ("memory:path/to/resource", ("memory",)), + ("memory:a", ("memory",)), + ("memory:another@path+with,lots&of(special;characters)*_!-$'", ("memory",)), + ("memory:/path/to/resource", ("memory",)), + ("memory://path/to/resource", ("memory",)), + # n5.md + ("file:///tmp/data.n5/|n5:path/to/array", ("file", "n5")), + ("file:///tmp/data.n5/|n5:a", ("file", "n5")), + ("file:///tmp/data.n5/|n5:/path/to/array", ("file", "n5")), + ("file:///tmp/data.n5/|n5:", ("file", "n5")), + ("file:///tmp/data.n5/|n5", ("file", "n5")), + # neuroglancer-precomputed.md + ( + "file:///tmp/dataset.precomputed/|neuroglancer-precomputed:", + ("file", "neuroglancer-precomputed"), + ), + ( + "file:///tmp/dataset.precomputed/|neuroglancer-precomputed", + ("file", "neuroglancer-precomputed"), + ), + # ocdbt.md + ("file:///path/to/repo.ocdbt/|ocdbt:", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo/", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:a", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt:/path/within/repo", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v123/", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v1", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://v123/path/within/repo", ("file", "ocdbt")), + ( + "file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database", + ("file", "ocdbt"), + ), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45Z", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.1Z", ("file", "ocdbt")), + ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.123456789Z", ("file", "ocdbt")), + # png.md + ("file:/path/to/image.png|png:", ("file", "png")), + ("file:/path/to/image.png|png", ("file", "png")), + # s3+http.md + ("s3+https://endpoint/path/within/bucket", ("s3+https",)), + ("s3+https://endpoint/bucket/path/within/bucket", ("s3+https",)), + ("s3+https://mybucket.s3.amazonaws.com/path/to/file", ("s3+https",)), + ("s3+https://s3.amazonaws.com/mybucket/path/to/file", ("s3+https",)), + ("s3+http://example.com", ("s3+http",)), + ("s3+http://example.com/", ("s3+http",)), + # s3.md + ("s3://bucket", ("s3",)), + ("s3://bucket/", ("s3",)), + ("s3://bucket/path/within/bucket", ("s3",)), + ("s3://aaa", ("s3",)), + ( + "s3://aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ("s3",), + ), + # tiff.md + ("file:/path/to/image.tiff|tiff:", ("file", "tiff")), + ("file:/path/to/image.tiff|tiff", ("file", "tiff")), + # webp.md + ("file:/path/to/image.webp|webp:", ("file", "webp")), + ("file:/path/to/image.webp|webp", ("file", "webp")), + # zarr.md + ("file:///path/to/node.zarr/|zarr3:", ("file", "zarr3")), + ("file:///path/to/node.zarr/|zarr3", ("file", "zarr3")), + ("file:///path/to/node.zarr/|zarr2:", ("file", "zarr2")), + ("file:///path/to/node.zarr/|zarr2", ("file", "zarr2")), + ("file:///path/to/node.zarr/|zarr:", ("file", "zarr")), + ("file:///path/to/node.zarr/|zarr", ("file", "zarr")), + # zip.md + ("file:/path/to/archive.zip|zip:path/to/file.txt", ("file", "zip")), + ("file:/path/to/archive.zip|zip", ("file", "zip")), + ("file:/path/to/archive.zip|zip:/path/to/file.txt", ("file", "zip")), + ("file:/path/to/outer.zip|zip:path/to/inner.zip|zip:path/to/file.txt", ("file", "zip", "zip")), + # zstd.md + ("file:/path/to/data.zstd|zstd:", ("file", "zstd")), + ("file:/path/to/data.zstd|zstd", ("file", "zstd")), +] + + +def _uppercase_schemes(example: str) -> str: + """Uppercase every sub-URL scheme, as the spec's linter does.""" + + def repl(match: re.Match[str]) -> str: + return match.group(1) + match.group(2).upper() + + return re.sub(r"((?:^|\|)\s*)([a-zA-Z][a-zA-Z0-9+.-]*)", repl, example) + + +@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=lambda v: str(v)) +def test_spec_example_parses(url: str, expected_schemes: tuple[str, ...]) -> None: + segments = parse_pipeline(url) + assert tuple(segment.scheme for segment in segments) == expected_schemes + # lossless round-trip of the original text + assert "|".join(segment.raw for segment in segments) == url + + +@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=lambda v: str(v)) +def test_spec_example_schemes_case_insensitive(url: str, expected_schemes: tuple[str, ...]) -> None: + upper = _uppercase_schemes(url) + segments = parse_pipeline(upper) + assert tuple(segment.scheme for segment in segments) == expected_schemes