feat: add HTTP store support

.github/workflows/cd.yml

@@ -34,6 +34,10 @@ jobs:
           - { os: linux, manylinux: musllinux_1_2, target: armv7 }
           # windows
           - { os: windows, target: i686, python-architecture: x86 }
+        exclude:
+          # https://github.com/rust-cross/cargo-xwin/issues/76
+          - os: windows
+            target: aarch64
     runs-on: ${{ (matrix.os == 'linux' && 'ubuntu') || matrix.os }}-latest
     steps:
       - uses: actions/checkout@v4

Cargo.toml

@@ -19,6 +19,9 @@ numpy = "0.23.0"
 unsafe_cell_slice = "0.2.0"
 serde_json = "1.0.128"
 pyo3-stub-gen = "0.6.1"
+opendal = { version = "0.50.2", features = ["services-http"] }
+tokio = { version = "1.41.1", features = ["rt-multi-thread"] }
+zarrs_opendal = "0.4.0"
 
 [profile.release]
 lto = true

README.md

@@ -30,7 +30,14 @@ You can then use your `zarr` as normal (with some caveats)!
 
 We export a `ZarrsCodecPipeline` class so that `zarr-python` can use the class but it is not meant to be instantiated and we do not guarantee the stability of its API beyond what is required so that `zarr-python` can use it.  Therefore, it is not documented here.  We also export two errors, `DiscontiguousArrayError` and `CollapsedDimensionError` that can be thrown in the process of converting to indexers that `zarrs` can understand (see below for more details).
 
-At the moment, we only support local filesystems but intend to support more in the future: https://github.com/ilan-gold/zarrs-python/issues/44
+At the moment, we only support a subset of the `zarr-python` stores:
+
+- [x] [LocalStore](https://zarr.readthedocs.io/en/main/_autoapi/zarr/storage/local/index.html) (FileSystem)
+- [RemoteStore](https://zarr.readthedocs.io/en/main/_autoapi/zarr/storage/remote/index.html)
+  - [x] [HTTPFileSystem](https://filesystem-spec.readthedocs.io/en/latest/api.html#fsspec.implementations.http.HTTPFileSystem)
+
+A `NotImplementedError` will be raised if a store is not supported.
+We intend to support more stores in the future: https://github.com/ilan-gold/zarrs-python/issues/44.
 
 ### Configuration
 

python/zarrs/_internal.pyi

@@ -2,6 +2,7 @@
 # ruff: noqa: E501, F401
 
 import typing
+from enum import Enum, auto
 
 import numpy
 import numpy.typing
@@ -21,7 +22,9 @@ class CodecPipelineImpl:
         self,
         chunk_descriptions: typing.Sequence[
             tuple[
-                tuple[str, typing.Sequence[int], str, typing.Sequence[int]],
+                tuple[
+                    StoreConfig, str, typing.Sequence[int], str, typing.Sequence[int]
+                ],
                 typing.Sequence[slice],
                 typing.Sequence[slice],
             ]
@@ -31,17 +34,29 @@ class CodecPipelineImpl:
     def retrieve_chunks(
         self,
         chunk_descriptions: typing.Sequence[
-            tuple[str, typing.Sequence[int], str, typing.Sequence[int]]
+            tuple[StoreConfig, str, typing.Sequence[int], str, typing.Sequence[int]]
         ],
     ) -> list[numpy.typing.NDArray[numpy.uint8]]: ...
     def store_chunks_with_indices(
         self,
         chunk_descriptions: typing.Sequence[
             tuple[
-                tuple[str, typing.Sequence[int], str, typing.Sequence[int]],
+                tuple[
+                    StoreConfig, str, typing.Sequence[int], str, typing.Sequence[int]
+                ],
                 typing.Sequence[slice],
                 typing.Sequence[slice],
             ]
         ],
         value: numpy.NDArray[typing.Any],
     ) -> None: ...
+
+class FilesystemStoreConfig:
+    root: str
+
+class HttpStoreConfig:
+    root: str
+
+class StoreConfig(Enum):
+    Filesystem = auto()
+    Http = auto()

python/zarrs/utils.py

@@ -12,7 +12,7 @@
     from collections.abc import Iterable
     from types import EllipsisType
 
-    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.common import ChunkCoords
 
@@ -63,10 +63,11 @@ def selector_tuple_to_slice_selection(selector_tuple: SelectorTuple) -> list[sli
 
 
 def convert_chunk_to_primitive(
-    byte_getter: ByteGetter | ByteSetter, chunk_spec: ArraySpec
-) -> tuple[str, ChunkCoords, str, Any]:
+    byte_interface: ByteGetter | ByteSetter, chunk_spec: ArraySpec
+) -> tuple[Store, str, ChunkCoords, str, Any]:
     return (
-        str(byte_getter),
+        byte_interface.store,
+        byte_interface.path,
         chunk_spec.shape,
         str(chunk_spec.dtype),
         chunk_spec.fill_value.tobytes(),
@@ -149,7 +150,7 @@ def make_chunk_info_for_rust_with_indices(
         tuple[ByteGetter | ByteSetter, ArraySpec, SelectorTuple, SelectorTuple]
     ],
     drop_axes: tuple[int, ...],
-) -> list[tuple[tuple[str, ChunkCoords, str, Any], list[slice], list[slice]]]:
+) -> list[tuple[tuple[Store, str, ChunkCoords, str, Any], list[slice], list[slice]]]:
     chunk_info_with_indices = []
     for byte_getter, chunk_spec, chunk_selection, out_selection in batch_info:
         chunk_info = convert_chunk_to_primitive(byte_getter, chunk_spec)
@@ -178,7 +179,7 @@ def make_chunk_info_for_rust(
     batch_info: Iterable[
         tuple[ByteGetter | ByteSetter, ArraySpec, SelectorTuple, SelectorTuple]
     ],
-) -> list[tuple[str, ChunkCoords, str, Any]]:
+) -> list[tuple[Store, str, ChunkCoords, str, Any]]:
     return list(
         convert_chunk_to_primitive(byte_getter, chunk_spec)
         for (byte_getter, chunk_spec, _, _) in batch_info

src/chunk_item.rs

@@ -1,4 +1,4 @@
-use std::{num::NonZeroU64, sync::Arc};
+use std::num::NonZeroU64;
 
 use pyo3::{
     exceptions::{PyRuntimeError, PyValueError},
@@ -9,13 +9,15 @@ use zarrs::{
     array::{ChunkRepresentation, DataType, FillValue},
     array_subset::ArraySubset,
     metadata::v3::{array::data_type::DataTypeMetadataV3, MetadataV3},
-    storage::{MaybeBytes, ReadableWritableListableStorageTraits, StorageError, StoreKey},
+    storage::{MaybeBytes, ReadableWritableListableStorage, StorageError, StoreKey},
 };
 
-use crate::utils::PyErrExt;
+use crate::{utils::PyErrExt, StoreConfig};
 
 pub(crate) type Raw<'a> = (
-    // store path
+    // store
+    StoreConfig,
+    // path
     String,
     // shape
     Vec<u64>,
@@ -34,17 +36,18 @@ pub(crate) type RawWithIndices<'a> = (
 );
 
 pub(crate) trait IntoItem<T, S>: std::marker::Sized {
-    fn store_path(&self) -> &str;
+    fn store_config(&self) -> &StoreConfig;
+    fn path(&self) -> &str;
     fn into_item(
         self,
-        store: Arc<dyn ReadableWritableListableStorageTraits>,
+        store: ReadableWritableListableStorage,
         key: StoreKey,
         shape: S,
     ) -> PyResult<T>;
 }
 
 pub(crate) trait ChunksItem {
-    fn store(&self) -> Arc<dyn ReadableWritableListableStorageTraits>;
+    fn store(&self) -> ReadableWritableListableStorage;
     fn key(&self) -> &StoreKey;
     fn representation(&self) -> &ChunkRepresentation;
 
@@ -54,7 +57,7 @@ pub(crate) trait ChunksItem {
 }
 
 pub(crate) struct Basic {
-    store: Arc<dyn ReadableWritableListableStorageTraits>,
+    store: ReadableWritableListableStorage,
     key: StoreKey,
     representation: ChunkRepresentation,
 }
@@ -66,7 +69,7 @@ pub(crate) struct WithSubset {
 }
 
 impl ChunksItem for Basic {
-    fn store(&self) -> Arc<dyn ReadableWritableListableStorageTraits> {
+    fn store(&self) -> ReadableWritableListableStorage {
         self.store.clone()
     }
     fn key(&self) -> &StoreKey {
@@ -78,7 +81,7 @@ impl ChunksItem for Basic {
 }
 
 impl ChunksItem for WithSubset {
-    fn store(&self) -> Arc<dyn ReadableWritableListableStorageTraits> {
+    fn store(&self) -> ReadableWritableListableStorage {
         self.item.store.clone()
     }
     fn key(&self) -> &StoreKey {
@@ -90,16 +93,21 @@ impl ChunksItem for WithSubset {
 }
 
 impl<'a> IntoItem<Basic, ()> for Raw<'a> {
-    fn store_path(&self) -> &str {
+    fn store_config(&self) -> &StoreConfig {
         &self.0
     }
+
+    fn path(&self) -> &str {
+        &self.1
+    }
+
     fn into_item(
         self,
-        store: Arc<dyn ReadableWritableListableStorageTraits>,
+        store: ReadableWritableListableStorage,
         key: StoreKey,
         (): (),
     ) -> PyResult<Basic> {
-        let (_, chunk_shape, dtype, fill_value) = self;
+        let (_, _, chunk_shape, dtype, fill_value) = self;
         let representation = get_chunk_representation(chunk_shape, &dtype, fill_value)?;
         Ok(Basic {
             store,
@@ -110,17 +118,22 @@ impl<'a> IntoItem<Basic, ()> for Raw<'a> {
 }
 
 impl IntoItem<WithSubset, &[u64]> for RawWithIndices<'_> {
-    fn store_path(&self) -> &str {
+    fn store_config(&self) -> &StoreConfig {
         &self.0 .0
     }
+
+    fn path(&self) -> &str {
+        &self.0 .1
+    }
+
     fn into_item(
         self,
-        store: Arc<dyn ReadableWritableListableStorageTraits>,
+        store: ReadableWritableListableStorage,
         key: StoreKey,
         shape: &[u64],
     ) -> PyResult<WithSubset> {
         let (raw, selection, chunk_selection) = self;
-        let chunk_shape = raw.1.clone();
+        let chunk_shape = raw.2.clone();
         let item = raw.into_item(store.clone(), key, ())?;
         Ok(WithSubset {
             item,

src/lib.rs

@@ -1,4 +1,5 @@
 #![warn(clippy::pedantic)]
+#![allow(clippy::module_name_repetitions)]
 
 use chunk_item::{ChunksItem, IntoItem};
 use concurrency::ChunkConcurrentLimitAndCodecOptions;
@@ -12,6 +13,7 @@ use rayon::iter::{IntoParallelIterator, ParallelIterator};
 use rayon_iter_concurrent_limit::iter_concurrent_limit;
 use std::borrow::Cow;
 use std::sync::{Arc, Mutex};
+use store::StoreConfig;
 use unsafe_cell_slice::UnsafeCellSlice;
 use zarrs::array::codec::{
     ArrayToBytesCodecTraits, CodecOptions, CodecOptionsBuilder, StoragePartialDecoder,
@@ -21,58 +23,46 @@ use zarrs::array::{
 };
 use zarrs::array_subset::ArraySubset;
 use zarrs::metadata::v3::MetadataV3;
-use zarrs::storage::{ReadableWritableListableStorageTraits, StorageHandle, StoreKey};
+use zarrs::storage::{ReadableWritableListableStorage, StorageHandle, StoreKey};
 
 mod chunk_item;
-mod codec_pipeline_store_filesystem;
 mod concurrency;
+mod runtime;
+mod store;
 #[cfg(test)]
 mod tests;
 mod utils;
 
-use codec_pipeline_store_filesystem::CodecPipelineStoreFilesystem;
 use utils::{PyErrExt, PyUntypedArrayExt};
 
-trait CodecPipelineStore: Send + Sync {
-    fn store(&self) -> Arc<dyn ReadableWritableListableStorageTraits>;
-    fn chunk_path(&self, store_path: &str) -> PyResult<String>;
-}
-
 // TODO: Use a OnceLock for store with get_or_try_init when stabilised?
 #[gen_stub_pyclass]
 #[pyclass]
 pub struct CodecPipelineImpl {
     pub(crate) codec_chain: Arc<CodecChain>,
-    pub(crate) store: Mutex<Option<Arc<dyn CodecPipelineStore>>>,
+    pub(crate) store: Mutex<Option<ReadableWritableListableStorage>>,
     pub(crate) codec_options: CodecOptions,
     pub(crate) chunk_concurrent_minimum: usize,
     pub(crate) chunk_concurrent_maximum: usize,
     pub(crate) num_threads: usize,
 }
 
 impl CodecPipelineImpl {
-    fn get_store_and_path(
+    fn get_store_from_config(
         &self,
-        store_path: &str,
-    ) -> PyResult<(Arc<dyn ReadableWritableListableStorageTraits>, String)> {
+        config: &StoreConfig,
+    ) -> PyResult<ReadableWritableListableStorage> {
         let mut gstore = self.store.lock().map_err(|_| {
             PyErr::new::<PyRuntimeError, _>("failed to lock the store mutex".to_string())
         })?;
 
-        #[allow(clippy::collapsible_if)]
-        if gstore.is_none() {
-            if store_path.starts_with("file://") {
-                *gstore = Some(Arc::new(CodecPipelineStoreFilesystem::new()?));
-            }
-            // TODO: Add support for more stores
-        }
-
+        // TODO: Request upstream change to get store on codec pipeline initialisation, do not want to do all of this here
         if let Some(gstore) = gstore.as_ref() {
-            Ok((gstore.store(), gstore.chunk_path(store_path)?))
+            Ok(gstore.clone())
         } else {
-            Err(PyErr::new::<PyTypeError, _>(format!(
-                "unsupported store for {store_path}"
-            )))
+            let store: ReadableWritableListableStorage = config.try_into()?;
+            *gstore = Some(store.clone());
+            Ok(store)
         }
     }
 
@@ -84,7 +74,9 @@ impl CodecPipelineImpl {
         chunk_descriptions
             .into_iter()
             .map(|raw| {
-                let (store, path) = self.get_store_and_path(raw.store_path())?;
+                // TODO: Prefer to get the store once, and assume it is the same for all chunks
+                let store = self.get_store_from_config(raw.store_config())?;
+                let path = raw.path();
                 let key = StoreKey::new(path).map_py_err::<PyValueError>()?;
                 raw.into_item(store, key, shape)
             })