feat: add from_array_metadata_and_store to CodecPipeline (#3233)

* feat: add `from_array_metadata_and_store` to `CodecPipeline`

This enables fancy `CodecPipeline` implementations like `zarrs-python` to support a broader range of Zarr arrays and to construct store wrappers on initialisation.

* make `store` keyword-only

Co-authored-by: Davis Bennett <davis.v.bennett@gmail.com>

* changelog

---------

Co-authored-by: Davis Bennett <davis.v.bennett@gmail.com>

Author: LDeakin

changes/3233.feature.rst

@@ -0,0 +1 @@
+Add an alternate `from_array_metadata_and_store` constructor to `CodecPipeline`.

src/zarr/abc/codec.py

@@ -12,11 +12,12 @@
     from collections.abc import Awaitable, Callable, Iterable
     from typing import Self
 
-    from zarr.abc.store import ByteGetter, ByteSetter
+    from zarr.abc.store import ByteGetter, ByteSetter, Store
     from zarr.core.array_spec import ArraySpec
     from zarr.core.chunk_grids import ChunkGrid
     from zarr.core.dtype.wrapper import TBaseDType, TBaseScalar, ZDType
     from zarr.core.indexing import SelectorTuple
+    from zarr.core.metadata import ArrayMetadata
 
 __all__ = [
     "ArrayArrayCodec",
@@ -281,6 +282,25 @@ def from_codecs(cls, codecs: Iterable[Codec]) -> Self:
         """
         ...
 
+    @classmethod
+    def from_array_metadata_and_store(cls, array_metadata: ArrayMetadata, store: Store) -> Self:
+        """Creates a codec pipeline from array metadata and a store path.
+
+        Raises NotImplementedError by default, indicating the CodecPipeline must be created with from_codecs instead.
+
+        Parameters
+        ----------
+        array_metadata : ArrayMetadata
+        store : Store
+
+        Returns
+        -------
+        Self
+        """
+        raise NotImplementedError(
+            f"'{type(cls).__name__}' does not implement CodecPipeline.from_array_metadata_and_store."
+        )
+
     @property
     @abstractmethod
     def supports_partial_decode(self) -> bool: ...

src/zarr/core/array.py

@@ -192,7 +192,15 @@ def parse_array_metadata(data: Any) -> ArrayMetadata:
     raise TypeError  # pragma: no cover
 
 
-def create_codec_pipeline(metadata: ArrayMetadata) -> CodecPipeline:
+def create_codec_pipeline(metadata: ArrayMetadata, *, store: Store | None = None) -> CodecPipeline:
+    if store is not None:
+        try:
+            return get_pipeline_class().from_array_metadata_and_store(
+                array_metadata=metadata, store=store
+            )
+        except NotImplementedError:
+            pass
+
     if isinstance(metadata, ArrayV3Metadata):
         return get_pipeline_class().from_codecs(metadata.codecs)
     elif isinstance(metadata, ArrayV2Metadata):
@@ -311,7 +319,11 @@ def __init__(
         object.__setattr__(self, "metadata", metadata_parsed)
         object.__setattr__(self, "store_path", store_path)
         object.__setattr__(self, "_config", config_parsed)
-        object.__setattr__(self, "codec_pipeline", create_codec_pipeline(metadata=metadata_parsed))
+        object.__setattr__(
+            self,
+            "codec_pipeline",
+            create_codec_pipeline(metadata=metadata_parsed, store=store_path.store),
+        )
 
     # this overload defines the function signature when zarr_format is 2
     @overload