diff --git a/changes/2943.feature.md b/changes/2943.feature.md new file mode 100644 index 0000000000..11779267b2 --- /dev/null +++ b/changes/2943.feature.md @@ -0,0 +1,9 @@ +Added core support for URL pipelines (https://github.com/jbms/url-pipeline): +`|`-chained URLs that address zarr data through nested storage layers, e.g. +`s3://bucket/data.zip|zip:|zarr3:`. This PR adds the parser, the single-method +`zarr.abc.url_pipeline.URLPipelineAdapter` interface, and the +`zarr.url_adapters` entry-point group through which third-party packages +(e.g. Icechunk) register adapters for their own schemes. Adapters for a scheme +are loaded lazily and individually; URLs without a `|` separator (and without +a registered root scheme) are handled exactly as before. Builtin adapters +(`zip:`, `zarr2:`/`zarr3:`) follow in separate pull requests. diff --git a/changes/4187.feature.md b/changes/4187.feature.md new file mode 100644 index 0000000000..87133e2034 --- /dev/null +++ b/changes/4187.feature.md @@ -0,0 +1,4 @@ +`ZipStore` now accepts an open binary file-like object in place of a path, enabling +zip archives on remote storage (e.g. a file opened with `fsspec` or an +`obstore.ReadableFile`). Operations that require a filesystem location +(`clear`, `move`) raise `NotImplementedError` for file-object-backed stores. diff --git a/docs/user-guide/storage.md b/docs/user-guide/storage.md index 7e0154b2a0..0ba6202c76 100644 --- a/docs/user-guide/storage.md +++ b/docs/user-guide/storage.md @@ -124,6 +124,19 @@ array = zarr.create_array(store=store, shape=(2,), dtype='float64') print(array) ``` +In place of a path, `ZipStore` also accepts an open binary file object (for +example a file opened with `fsspec`, or an `obstore` reader), enabling zip +archives on remote storage. The file must stay open for as long as the store +is in use: + +```python exec="true" session="storage" source="above" result="ansi" +store.close() +f = open('data.zip', mode='rb') # must stay open while the store is used +array = zarr.open_array(store=zarr.storage.ZipStore(f), mode='r') +print(array[:]) +f.close() +``` + ### Remote Store The [`zarr.storage.FsspecStore`][] stores the contents of a Zarr hierarchy following the same diff --git a/src/zarr/abc/url_pipeline.py b/src/zarr/abc/url_pipeline.py new file mode 100644 index 0000000000..56202cbd10 --- /dev/null +++ b/src/zarr/abc/url_pipeline.py @@ -0,0 +1,174 @@ +""" +Abstract base class and data model for URL pipeline adapters. + +A URL pipeline is a `|`-separated chain of sub-URLs, read outer-to-inner, +as specified by https://github.com/jbms/url-pipeline. The first sub-URL (the +*root*) locates a resource using a conventional URL, and each subsequent +sub-URL names an *adapter* that reinterprets everything to its left: + + s3://bucket/data.zip|zip:path/inside|zarr3: + +Third-party packages provide adapters by subclassing +[`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] and +registering the class under the `zarr.url_adapters` entry-point group, +using the URL scheme as the entry-point name. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + + from zarr.abc.store import Store + from zarr.core.common import AccessModeLiteral, ZarrFormat + +__all__ = [ + "AdapterResolution", + "PipelineContext", + "PipelineSegment", + "URLPipelineAdapter", +] + + +@dataclass(frozen=True) +class PipelineSegment: + """ + One `|`-delimited sub-URL of a URL pipeline. + + Attributes + ---------- + scheme : str + The lowercased URL scheme. Empty string only for a schemeless root + (a bare local path). + body : str + The text after `scheme:` and before any `?`. Interpretation is + scheme-defined; it is **not** URL-normalized, so case-significant + content (e.g. icechunk snapshot IDs) is preserved. + query : str | None + The raw query string after `?`, or None. Interpretation is + scheme-defined. + raw : str + The exact original sub-URL text, preserved for lossless + reconstruction of the pipeline. + """ + + scheme: str + body: str + query: str | None + raw: str + + def __str__(self) -> str: + return self.raw + + +@dataclass(frozen=True) +class AdapterResolution: + """ + The result of resolving a URL pipeline (or a prefix of one). + + Attributes + ---------- + store : Store + The resolved store. + path : str + Residual path *within* the store that the pipeline addresses + (e.g. `"path/to/node"` for `...|icechunk://tag.v1/path/to/node`). + Empty string when the pipeline addresses the store root. + zarr_format : ZarrFormat | None + Zarr format selected by a format segment (`zarr2:`/`zarr3:`), + or None if unspecified. + """ + + store: Store + path: str = "" + zarr_format: ZarrFormat | None = None + + +@dataclass(frozen=True) +class PipelineContext: + """ + Context handed to a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] + describing the pipeline to the left of its segment. + + Attributes + ---------- + preceding : tuple[PipelineSegment, ...] + The parsed sub-URLs to the left of the adapter's segment, outer to + inner. Empty when the adapter's segment is the pipeline root. + mode : AccessModeLiteral | None + The access mode requested by the caller (e.g. `zarr.open(mode=...)`), + or None when unspecified. Adapters for read-only resources should + raise for explicit write modes (`"w"`, `"w-"`, `"a"`, `"r+"`) + and open read-only for `None` and `"r"`. + read_only : bool + True when `mode == "r"`. Adapters must construct their store + read-only when this is set. + storage_options : dict[str, Any] | None + Options passed by the caller. By convention these configure the + *root* sub-URL (e.g. fsspec options), but adapters may consume + adapter-specific keys. + """ + + preceding: tuple[PipelineSegment, ...] + mode: AccessModeLiteral | None + read_only: bool + storage_options: dict[str, Any] | None + _resolver: Callable[[tuple[PipelineSegment, ...]], Awaitable[AdapterResolution]] = field( + repr=False + ) + + @property + def preceding_url(self) -> str: + """The pipeline to the left of this segment, reconstructed exactly.""" + return "|".join(segment.raw for segment in self.preceding) + + async def resolve_preceding(self) -> AdapterResolution: + """ + Resolve the preceding pipeline into a store. + + This is the entry point for *wrapper* adapters (e.g. `zip:`) that + operate on the resource produced by the segments to their left. + Adapters backed by their own I/O machinery (e.g. `icechunk:`) + should use [`preceding_url`][zarr.abc.url_pipeline.PipelineContext.preceding_url] + instead and never materialize the intermediate store. + """ + return await self._resolver(self.preceding) + + +class URLPipelineAdapter(ABC): + """ + Handler for one URL pipeline scheme. + + Subclasses implement a single classmethod, + [`open_pipeline_segment`][zarr.abc.url_pipeline.URLPipelineAdapter.open_pipeline_segment], + and are registered under the `zarr.url_adapters` entry-point group with + the URL scheme as the entry-point name: + + [project.entry-points."zarr.url_adapters"] + myscheme = "mypackage.zarr_adapter:MyAdapter" + + An adapter is used in two positions: + + - as an *adapter segment*: `s3://bucket/repo|icechunk://tag.v1` — the + context carries the preceding sub-URLs; + - as a *root scheme*: `gh://org/repo` — `context.preceding` is empty. + """ + + @classmethod + @abstractmethod + async def open_pipeline_segment( + cls, segment: PipelineSegment, context: PipelineContext + ) -> AdapterResolution: + """ + Resolve `segment` (in the context of the pipeline to its left) + into a store, an optional residual path within that store, and an + optional zarr format. + + The returned store must already be open and must honor + `context.read_only`. + """ + ... 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 48f60fabd7..bf598231ba 100644 --- a/src/zarr/registry.py +++ b/src/zarr/registry.py @@ -21,6 +21,7 @@ CodecPipeline, ) from zarr.abc.numcodec import Numcodec + from zarr.abc.url_pipeline import URLPipelineAdapter from zarr.core.buffer import Buffer, NDBuffer from zarr.core.chunk_key_encodings import ChunkKeyEncoding from zarr.core.common import JSON @@ -32,11 +33,14 @@ "get_codec_class", "get_ndbuffer_class", "get_pipeline_class", + "get_url_adapter", + "list_url_adapter_schemes", "register_buffer", "register_chunk_key_encoding", "register_codec", "register_ndbuffer", "register_pipeline", + "register_url_adapter", ] @@ -62,6 +66,7 @@ def register(self, cls: type[T], qualname: str | None = None) -> None: _buffer_registry: Registry[Buffer] = Registry() _ndbuffer_registry: Registry[NDBuffer] = Registry() _chunk_key_encoding_registry: Registry[ChunkKeyEncoding] = Registry() +_url_adapter_registry: Registry[URLPipelineAdapter] = Registry() """ The registry module is responsible for managing implementations of codecs, @@ -108,6 +113,8 @@ def _collect_entrypoints() -> list[Registry[Any]]: entry_points.select(group="zarr", name="chunk_key_encoding") ) + _url_adapter_registry.lazy_load_list.extend(entry_points.select(group="zarr.url_adapters")) + _pipeline_registry.lazy_load_list.extend(entry_points.select(group="zarr.codec_pipeline")) _pipeline_registry.lazy_load_list.extend( entry_points.select(group="zarr", name="codec_pipeline") @@ -124,6 +131,7 @@ def _collect_entrypoints() -> list[Registry[Any]]: _buffer_registry, _ndbuffer_registry, _chunk_key_encoding_registry, + _url_adapter_registry, ] @@ -303,6 +311,53 @@ def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]: return _chunk_key_encoding_registry[key] +def register_url_adapter(scheme: str, cls: type[URLPipelineAdapter]) -> None: + """ + Register a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] + class for a URL scheme. + """ + _url_adapter_registry.register(cls, scheme.lower()) + + +def list_url_adapter_schemes() -> set[str]: + """ + The set of URL schemes with a registered URL pipeline adapter. + + Includes adapters advertised via not-yet-loaded `zarr.url_adapters` + entry points; consulting this does not import any adapter code. + """ + return set(_url_adapter_registry) | {e.name for e in _url_adapter_registry.lazy_load_list} + + +def get_url_adapter(scheme: str) -> type[URLPipelineAdapter]: + """ + Get the URL pipeline adapter class registered for `scheme`. + + Loads pending `zarr.url_adapters` entry points for this scheme only, so + resolving one scheme never imports other providers' packages. + """ + from zarr.errors import URLPipelineError + + key = scheme.lower() + if key not in _url_adapter_registry: + remaining = [] + for entry_point in _url_adapter_registry.lazy_load_list: + if entry_point.name == key: + _url_adapter_registry.register(entry_point.load(), qualname=key) + else: + remaining.append(entry_point) + _url_adapter_registry.lazy_load_list[:] = remaining + try: + return _url_adapter_registry[key] + except KeyError: + registered = sorted(list_url_adapter_schemes()) + raise URLPipelineError( + f"no URL pipeline adapter is registered for scheme {scheme!r}. " + f"Registered schemes: {registered}. Adapters are provided by " + "packages via the 'zarr.url_adapters' entry-point group." + ) from None + + _collect_entrypoints() diff --git a/src/zarr/storage/_common.py b/src/zarr/storage/_common.py index 7e9c035c69..e3d985280e 100644 --- a/src/zarr/storage/_common.py +++ b/src/zarr/storage/_common.py @@ -21,11 +21,24 @@ 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._utils import _join_paths, normalize_path, parse_store_url + +def _is_url_pipeline(url: str) -> bool: + """Whether a string store should be routed through the URL pipeline machinery.""" + from zarr.storage._url_pipeline import is_url_pipeline + + return is_url_pipeline(url) + + _has_fsspec = importlib.util.find_spec("fsspec") if _has_fsspec: from fsspec.mapping import FSMap @@ -47,14 +60,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: @@ -348,6 +369,17 @@ async def make_store( """ from zarr.storage._fsspec import FsspecStore # circular import + if isinstance(store_like, str) and _is_url_pipeline(store_like): + from zarr.storage._url_pipeline import resolve_pipeline + + result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options) + if result.path: + raise URLPipelineError( + f"the URL pipeline {store_like!r} resolves to a path inside a store; " + "use zarr.open() or make_store_path() instead of make_store()" + ) + return result.store + # Parse URL early so we can reuse the result for both validation and routing parsed = parse_store_url(store_like) if isinstance(store_like, str) else None @@ -467,6 +499,15 @@ async def make_store_path( "'path' was provided but is not used for FSMap store_like objects. Specify the path when creating the FSMap instance instead." ) + elif isinstance(store_like, str) and _is_url_pipeline(store_like): + from zarr.storage._url_pipeline import resolve_pipeline + + 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 + else: store = await make_store(store_like, mode=mode, storage_options=storage_options) return await StorePath.open(store, path=path_normalized, mode=mode) diff --git a/src/zarr/storage/_url_pipeline.py b/src/zarr/storage/_url_pipeline.py new file mode 100644 index 0000000000..9c40cb2751 --- /dev/null +++ b/src/zarr/storage/_url_pipeline.py @@ -0,0 +1,174 @@ +""" +Parsing and resolution of URL pipelines (https://github.com/jbms/url-pipeline). + +This module is imported lazily from `zarr.storage._common` — importing zarr +does not import it, and no adapters are loaded until a pipeline URL is +actually resolved. +""" + +from __future__ import annotations + +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._utils import parse_store_url + +if TYPE_CHECKING: + 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+.\-]*$") + + +def _root_scheme(root_sub_url: str) -> str: + """ + Detect the scheme of the root sub-URL. + + Uses `parse_store_url` (which knows Windows drive letters are not + schemes), falling back to plain scheme extraction when `urlparse` + rejects an exotic authority (e.g. bracketed non-IP text) — the pipeline + grammar constrains only the scheme, not the authority. + """ + try: + return parse_store_url(root_sub_url).scheme.lower() + except ValueError: + match = re.match(r"([a-zA-Z][a-zA-Z0-9+.\-]*):", root_sub_url) + return match.group(1).lower() if match else "" + + +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:`). + + 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): + body_and_scheme, query = _split_query(part) + if index == 0: + scheme = _root_scheme(body_and_scheme) + body = body_and_scheme + if scheme and body_and_scheme.lower().startswith(f"{scheme}:"): + body = body_and_scheme[len(scheme) + 1 :] + segments.append(PipelineSegment(scheme=scheme, body=body, query=query, raw=part)) + else: + 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 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:`). + The registry check inspects entry-point names only — no adapter code is + imported here. + """ + if "|" in url: + return True + scheme = parse_store_url(url).scheme.lower() + return bool(scheme) and scheme in list_url_adapter_schemes() + + +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, residual path, and optional + zarr format. + + Parameters + ---------- + url : str + A URL pipeline, e.g. `"s3://bucket/data.zip|zip:|zarr3:"`. + mode : AccessModeLiteral | None + The caller's access mode. `"r"` requires adapters to construct + read-only stores. + storage_options : dict | None + Options forwarded to the root sub-URL's store (and visible to + adapters via the context). + """ + segments = parse_pipeline(url) + if len(segments) == 1 and segments[0].scheme not in list_url_adapter_schemes(): + 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 segments[0].scheme not in list_url_adapter_schemes(): + # Base case: a plain root URL. Delegate to the existing StoreLike + # machinery (local paths, memory://, fsspec fallthrough) unchanged. + from zarr.storage._common import make_store + + store = await make_store(segments[0].raw, mode=mode, storage_options=storage_options) + return AdapterResolution(store=store) + + *preceding, last = segments + adapter_cls = get_url_adapter(last.scheme) + + async def _resolver(preceding: tuple[PipelineSegment, ...]) -> AdapterResolution: + if not preceding: + raise URLPipelineError( + f"adapter {last.scheme!r} was used as a pipeline root but " + "requires a preceding sub-URL" + ) + return await _resolve(preceding, mode=mode, storage_options=storage_options) + + context = PipelineContext( + preceding=tuple(preceding), + mode=mode, + read_only=mode == "r", + storage_options=storage_options, + _resolver=_resolver, + ) + return await adapter_cls.open_pipeline_segment(last, context) diff --git a/src/zarr/storage/_zip.py b/src/zarr/storage/_zip.py index 430b0c3e2a..69ae18bc2c 100644 --- a/src/zarr/storage/_zip.py +++ b/src/zarr/storage/_zip.py @@ -1,12 +1,13 @@ from __future__ import annotations +import io import os import shutil import threading import time import zipfile from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal +from typing import IO, TYPE_CHECKING, Any, Literal from zarr.abc.store import ( ByteRequest, @@ -23,14 +24,67 @@ ZipStoreAccessModeLiteral = Literal["r", "w", "a"] +class _RawReaderAdapter(io.RawIOBase): + """ + Adapt a minimal seekable reader to the `io` interface `zipfile` needs. + + Some file-like objects (e.g. `obstore.ReadableFile`) implement + `read`/`seek`/`tell` but are not `io.IOBase` instances, and their + `read` may return a buffer-protocol object rather than `bytes`. + Wrapping in this adapter plus `io.BufferedReader` yields real `bytes`. + + Reads are clamped to the bytes remaining before EOF: some readers + (obstore < 0.6) raise on short reads rather than returning fewer bytes. + The size is cached, which is safe because the adapter is only used for + read-only access. + """ + + def __init__(self, fileobj: IO[bytes]) -> None: + self._fileobj = fileobj + self._size: int | None = None + + def _get_size(self) -> int: + if self._size is None: + pos = self._fileobj.tell() + self._size = self._fileobj.seek(0, os.SEEK_END) + self._fileobj.seek(pos) + return self._size + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + def seek(self, pos: int, whence: int = 0) -> int: + return self._fileobj.seek(pos, whence) + + def tell(self) -> int: + return self._fileobj.tell() + + def readinto(self, b: Any) -> int: + n_requested = min(len(b), self._get_size() - self._fileobj.tell()) + if n_requested <= 0: + return 0 + data = self._fileobj.read(n_requested) + n = len(data) + b[:n] = memoryview(data) + return n + + class ZipStore(Store): """ Store using a ZIP file. Parameters ---------- - path : str - Location of file. + path : str, Path, or IO[bytes] + Location of file, or an open binary file object. A file object must + support `read`, `seek`, and `tell`; objects that are not `io.IOBase` + instances (e.g. an `obstore` reader) are adapted automatically but + can only be used for reading (`mode="r"`). The file object must stay + open for the lifetime of the store, and operations that require a + filesystem location (`clear`, `move`, pickling) are not supported. mode : str, optional One of 'r' to read an existing file, 'w' to truncate and write a new file, 'a' to append to an existing file, or 'x' to exclusively create @@ -58,16 +112,17 @@ class ZipStore(Store): supports_deletes: bool = False supports_listing: bool = True - path: Path + path: Path | None compression: int allowZip64: bool _zf: zipfile.ZipFile _lock: threading.RLock + _fileobj: IO[bytes] | None def __init__( self, - path: Path | str, + path: Path | str | IO[bytes], *, mode: ZipStoreAccessModeLiteral = "r", read_only: bool | None = None, @@ -81,8 +136,28 @@ def __init__( if isinstance(path, str): path = Path(path) - assert isinstance(path, Path) - self.path = path # root? + if isinstance(path, Path): + self.path = path # root? + self._fileobj = None + else: + self.path = None + if not isinstance(path, io.IOBase): + if not all( + callable(getattr(path, attr, None)) for attr in ("read", "seek", "tell") + ): + raise TypeError( + f"expected a path or an open binary file object supporting " + f"read/seek/tell, got {type(path).__name__}" + ) + if mode != "r": + raise TypeError( + f"a file object that is not an io.IOBase instance can only be " + f"opened for reading (mode='r', got mode={mode!r})" + ) + # e.g. an obstore ReadableFile: readable and seekable, but + # not an io object and reads may not return bytes + path = io.BufferedReader(_RawReaderAdapter(path)) + self._fileobj = path self._zmode = mode self.compression = compression @@ -95,7 +170,7 @@ def _sync_open(self) -> None: self._lock = threading.RLock() self._zf = zipfile.ZipFile( - self.path, + self.path if self.path is not None else self._fileobj, # type: ignore[arg-type] mode=self._zmode, compression=self.compression, allowZip64=self.allowZip64, @@ -107,6 +182,13 @@ async def _open(self) -> None: self._sync_open() def __getstate__(self) -> dict[str, Any]: + if self.path is None: + # A path-backed store pickles its path and reopens the file on + # unpickling; an open file object cannot be serialized that way. + raise TypeError( + "cannot pickle a ZipStore backed by a file-like object; " + "construct the store from a path instead" + ) # We need a copy to not modify the state of the original store state = self.__dict__.copy() for attr in ["_zf", "_lock"]: @@ -130,6 +212,10 @@ async def clear(self) -> None: # docstring inherited with self._lock: self._check_writable() + if self.path is None: + raise NotImplementedError( + "clear() is not supported for a ZipStore backed by a file-like object" + ) self._zf.close() os.remove(self.path) self._zf = zipfile.ZipFile( @@ -137,13 +223,19 @@ async def clear(self) -> None: ) def __str__(self) -> str: + if self.path is None: + return f"zip://{self._fileobj!r}" return f"zip://{self.path}" def __repr__(self) -> str: return f"ZipStore('{self}')" def __eq__(self, other: object) -> bool: - return isinstance(other, type(self)) and self.path == other.path + return ( + isinstance(other, type(self)) + and self.path == other.path + and self._fileobj is other._fileobj + ) def _get( self, @@ -297,6 +389,10 @@ async def move(self, path: Path | str) -> None: """ Move the store to another path. """ + if self.path is None: + raise NotImplementedError( + "move() is not supported for a ZipStore backed by a file-like object" + ) if isinstance(path, str): path = Path(path) self.close() 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..c7bb830901 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] +entrypoint-scheme = package_with_entrypoint:TestEntrypointURLAdapter diff --git a/tests/package_with_entrypoint/__init__.py b/tests/package_with_entrypoint/__init__.py index 23afcf1dc2..b128b8cca1 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=False) + return AdapterResolution(store=store, path=segment.body) diff --git a/tests/test_store/test_zip.py b/tests/test_store/test_zip.py index ed69114b51..0d8dadd18a 100644 --- a/tests/test_store/test_zip.py +++ b/tests/test_store/test_zip.py @@ -1,6 +1,8 @@ from __future__ import annotations +import io import os +import pickle import shutil import tempfile import zipfile @@ -188,6 +190,163 @@ async def test_move(self, tmp_path: Path) -> None: assert np.array_equal(array[...], np.arange(10)) +class TestZipStoreFileObj: + """ZipStore backed by an open binary file-like object instead of a path.""" + + @pytest.fixture + def zip_bytes(self, tmp_path: Path) -> bytes: + path = tmp_path / "data.zip" + store = ZipStore(path, mode="w") + zarr.create_array(store, data=np.arange(10), chunks=(5,)) + store.close() + return path.read_bytes() + + def test_read_from_fileobj(self, zip_bytes: bytes) -> None: + # an existing archive can be read through any seekable binary reader + store = ZipStore(io.BytesIO(zip_bytes), mode="r") + array = zarr.open_array(store, mode="r") + assert np.array_equal(array[...], np.arange(10)) + assert store.path is None + + def test_write_to_fileobj(self) -> None: + # a writable file object receives the archive; the bytes it holds + # after close() are a complete, reopenable zip + buffer = io.BytesIO() + store = ZipStore(buffer, mode="w", read_only=False) + zarr.create_array(store, data=np.arange(4)) + store.close() + + roundtrip = ZipStore(io.BytesIO(buffer.getvalue()), mode="r") + array = zarr.open_array(roundtrip, mode="r") + assert np.array_equal(array[...], np.arange(4)) + + async def test_clear_unsupported(self, zip_bytes: bytes) -> None: + # clear() requires a filesystem location, so it raises a clear error + # for file-object-backed stores + store = ZipStore(io.BytesIO(zip_bytes), mode="a", read_only=False) + store._sync_open() + with pytest.raises(NotImplementedError, match="clear.*file-like"): + await store.clear() + + async def test_move_unsupported(self, zip_bytes: bytes) -> None: + # move() requires a filesystem location, so it raises a clear error + # for file-object-backed stores + store = ZipStore(io.BytesIO(zip_bytes), mode="a", read_only=False) + store._sync_open() + with pytest.raises(NotImplementedError, match="move.*file-like"): + await store.move("elsewhere.zip") + + def test_invalid_file_object_rejected(self) -> None: + # objects without read/seek/tell are rejected at construction, not + # deep inside zipfile + with pytest.raises(TypeError, match="read/seek/tell"): + ZipStore(42, mode="r") # type: ignore[arg-type] + + @pytest.mark.parametrize("mode", ["w", "a", "x"]) + def test_non_iobase_reader_write_modes_rejected(self, zip_bytes: bytes, mode: str) -> None: + # readers that are not io.IOBase instances are adapted for reading + # only; write modes are rejected at construction with a clear error + class MinimalReader: + def __init__(self, data: bytes) -> None: + self._buffer = io.BytesIO(data) + + def read(self, size: int, /) -> bytes: + return self._buffer.read(size) + + def seek(self, pos: int, whence: int = 0, /) -> int: + return self._buffer.seek(pos, whence) + + def tell(self) -> int: + return self._buffer.tell() + + with pytest.raises(TypeError, match="opened for reading"): + ZipStore(MinimalReader(zip_bytes), mode=mode, read_only=False) # type: ignore[arg-type] + + def test_fsspec_file(self, tmp_path: Path, zip_bytes: bytes) -> None: + # a file opened through fsspec (already an io.IOBase) is used directly; + # fsspec's local filesystem stands in for a remote one + fsspec = pytest.importorskip("fsspec") + + path = tmp_path / "fsspec.zip" + path.write_bytes(zip_bytes) + with fsspec.open(f"local://{path}", "rb") as fileobj: + store = ZipStore(fileobj, mode="r") + array = zarr.open_array(store, mode="r") + assert np.array_equal(array[...], np.arange(10)) + assert store.path is None + + def test_obstore_reader(self, tmp_path: Path, zip_bytes: bytes) -> None: + # obstore's ReadableFile is not an io.IOBase and its read() returns a + # buffer-protocol object; ZipStore adapts it via _RawReaderAdapter + obstore = pytest.importorskip("obstore") + from obstore.store import LocalStore as ObstoreLocalStore + + (tmp_path / "obstore.zip").write_bytes(zip_bytes) + reader = obstore.open_reader(ObstoreLocalStore(str(tmp_path)), "obstore.zip") + store = ZipStore(reader, mode="r") + array = zarr.open_array(store, mode="r") + assert np.array_equal(array[...], np.arange(10)) + + def test_raw_reader_adapter_eof(self) -> None: + from zarr.storage._zip import _RawReaderAdapter + + class MinimalReader: + """Non-io.IOBase reader exposing only read/seek/tell, like obstore.""" + + def __init__(self, data: bytes) -> None: + self._buffer = io.BytesIO(data) + + def read(self, size: int, /) -> bytes: + return self._buffer.read(size) + + def seek(self, pos: int, whence: int = 0, /) -> int: + return self._buffer.seek(pos, whence) + + def tell(self) -> int: + return self._buffer.tell() + + # the adapter must clamp reads to EOF: some readers (obstore < 0.6) + # raise on short reads instead of returning fewer bytes + data = b"0123456789" + adapter = _RawReaderAdapter(MinimalReader(data)) # type: ignore[arg-type] + + # A read straddling EOF returns only the remaining bytes. + adapter.seek(len(data) - 3) + buf = bytearray(8) + assert adapter.readinto(buf) == 3 + assert bytes(buf[:3]) == data[-3:] + + # A read at EOF returns 0. + assert adapter.tell() == len(data) + assert adapter.readinto(bytearray(8)) == 0 + + def test_pickle_fileobj_raises(self, zip_bytes: bytes) -> None: + # an open file object cannot be reliably serialized, so pickling a + # file-object-backed store raises with a pointer at the alternative + store = ZipStore(io.BytesIO(zip_bytes), mode="r") + with pytest.raises(TypeError, match="cannot pickle a ZipStore backed by a file-like"): + pickle.dumps(store) + + def test_pickle_path_backed_roundtrip(self, tmp_path: Path, zip_bytes: bytes) -> None: + # path-backed stores remain picklable: the path is serialized and the + # archive is reopened on unpickling + path = tmp_path / "pickled.zip" + path.write_bytes(zip_bytes) + store = ZipStore(path, mode="r") + unpickled = pickle.loads(pickle.dumps(store)) + array = zarr.open_array(unpickled, mode="r") + assert np.array_equal(array[...], np.arange(10)) + + def test_str_and_eq(self, zip_bytes: bytes) -> None: + # file-object-backed stores stringify with the object repr and + # compare equal only when backed by the very same file object + fileobj = io.BytesIO(zip_bytes) + store = ZipStore(fileobj, mode="r") + assert str(store).startswith("zip://<") + assert store == ZipStore(fileobj, mode="r") + assert store != ZipStore(io.BytesIO(zip_bytes), mode="r") + + class ZipStoreLifecycleMachine(RuleBasedStateMachine): """Drive a ZipStore through construct / open / write / close transitions. 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..6c6f76464e --- /dev/null +++ b/tests/test_url_pipeline/test_parser.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import pytest + +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" + + +@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_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 diff --git a/tests/test_url_pipeline/test_resolver.py b/tests/test_url_pipeline/test_resolver.py new file mode 100644 index 0000000000..4e9601a7da --- /dev/null +++ b/tests/test_url_pipeline/test_resolver.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, ClassVar + +import pytest + +import zarr +import zarr.registry +from zarr.abc.url_pipeline import ( + AdapterResolution, + PipelineContext, + PipelineSegment, + URLPipelineAdapter, +) +from zarr.errors import URLPipelineError +from zarr.registry import ( + get_url_adapter, + list_url_adapter_schemes, + register_url_adapter, +) +from zarr.storage import MemoryStore, WrapperStore +from zarr.storage._common import make_store, make_store_path +from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline + +if TYPE_CHECKING: + from zarr.abc.store import Store + +pytestmark = pytest.mark.usefixtures("clean_url_adapter_registry") + + +class TracingStore(WrapperStore[MemoryStore]): + """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.""" + + @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 AdapterResolution(store=store, 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-", "a", "r+"): + raise ValueError("read-only scheme") + store = await MemoryStore.open(read_only=True) + return AdapterResolution(store=store, path=segment.body.lstrip("/")) + + +async def _store_of(resolution: AdapterResolution) -> Store: + return resolution.store + + +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") + + @pytest.mark.usefixtures("set_path") + def test_entrypoint_discovery(self) -> None: + assert "entrypoint-scheme" in list_url_adapter_schemes() + cls = get_url_adapter("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|entrypoint-scheme:sub/path") + assert result.path == "sub/path" + + @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 == "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 == "entrypoint-scheme" for e in registry.lazy_load_list) + assert get_url_adapter("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) + + +class TestResolve: + async def test_wrapper_adapter_chain(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline("memory://base|wrap:inner/path") + assert isinstance(result.store, TracingStore) + assert result.path == "inner/path" + assert result.store.context.preceding_url == "memory://base" + assert result.zarr_format is None + + 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) -> None: + register_url_adapter("wrap", WrapperAdapter) + result = await resolve_pipeline("memory://base|wrap:", mode="r") + assert result.store.context.read_only + assert result.store.context.mode == "r" + result = await resolve_pipeline("memory://base|wrap:") + assert not result.store.context.read_only + assert result.store.context.mode is None + + 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="requires a 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:") + + +class TestMakeStoreIntegration: + async def test_make_store_path_combines_paths(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + store_path = await make_store_path("memory://base|wrap:residual", path="user/sub") + assert store_path.path == "residual/user/sub" + assert store_path.zarr_format is None + + async def test_make_store_rejects_residual_path(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + with pytest.raises(URLPipelineError, match="resolves to a path inside a store"): + await make_store("memory://base|wrap:residual") + + async def test_make_store_no_residual_path(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + store = await make_store("memory://base|wrap:") + assert isinstance(store, TracingStore) + + async def test_zarr_open_end_to_end(self) -> None: + register_url_adapter("wrap", WrapperAdapter) + group = zarr.open_group("memory://base|wrap:", mode="w") + array = group.create_array("x", shape=(4,), dtype="i4") + array[:] = [1, 2, 3, 4] + assert zarr.open_array("memory://base|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_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}) 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