Merge branch 'main' into fix/chunkkeyencodinglike

Author: d-v-b

.gitignore

@@ -83,6 +83,7 @@ src/zarr/_version.py
 data/*
 src/fixture/
 fixture/
+junit.xml
 
 .DS_Store
 tests/.hypothesis

changes/2693.bugfix.rst

@@ -0,0 +1,13 @@
+Implement open() for LoggingStore
+LoggingStore is now a generic class.
+Use stdout rather than stderr as the default stream for LoggingStore
+Ensure that ZipStore is open before getting or setting any values
+Update equality for LoggingStore and WrapperStore such that 'other' must also be a LoggingStore or WrapperStore respectively, rather than only checking the types of the stores they wrap.
+Indicate StoreTest's `test_store_repr`, `test_store_supports_writes`, `test_store_supports_partial_writes`, and `test_store_supports_listing` need to be implemented using `@abstractmethod` rather than `NotImplementedError`.
+Separate instantiating and opening a store in StoreTests
+Test using Store as a context manager in StoreTests
+Match the errors raised by read only stores in StoreTests
+Test that a ValueError is raise for invalid byte range syntax in StoreTests
+Test getsize() and getsize_prefix() in StoreTests
+Test the error raised for invalid buffer arguments in StoreTests
+Test that data can be written to a store that's not yet open using the store.set method in StoreTests

changes/2762.bugfix.rst

@@ -0,0 +1,2 @@
+Fixed ZipStore to make sure the correct attributes are saved when instances are pickled.
+This fixes a previous bug that prevent using ZipStore with a ProcessPoolExecutor.

src/zarr/abc/store.py

@@ -176,10 +176,10 @@ async def get(
         Parameters
         ----------
         key : str
+        prototype : BufferPrototype
+            The prototype of the output buffer. Stores may support a default buffer prototype.
         byte_range : ByteRequest, optional
-
             ByteRequest may be one of the following. If not provided, all data associated with the key is retrieved.
-
             - RangeByteRequest(int, int): Request a specific range of bytes in the form (start, end). The end is exclusive. If the given range is zero-length or starts after the end of the object, an error will be returned. Additionally, if the range ends after the end of the object, the entire remainder of the object will be returned. Otherwise, the exact requested range will be returned.
             - OffsetByteRequest(int): Request all bytes starting from a given byte offset. This is equivalent to bytes={int}- as an HTTP header.
             - SuffixByteRequest(int): Request the last int bytes. Note that here, int is the size of the request, not the byte offset. This is equivalent to bytes=-{int} as an HTTP header.
@@ -200,6 +200,8 @@ async def get_partial_values(
 
         Parameters
         ----------
+        prototype : BufferPrototype
+            The prototype of the output buffer. Stores may support a default buffer prototype.
         key_ranges : Iterable[tuple[str, tuple[int | None, int | None]]]
             Ordered set of key, range pairs, a key may occur multiple times with different ranges
 

src/zarr/storage/_fsspec.py

@@ -10,14 +10,15 @@
     Store,
     SuffixByteRequest,
 )
+from zarr.core.buffer import Buffer
 from zarr.storage._common import _dereference_path
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable
 
     from fsspec.asyn import AsyncFileSystem
 
-    from zarr.core.buffer import Buffer, BufferPrototype
+    from zarr.core.buffer import BufferPrototype
     from zarr.core.common import BytesLike
 
 
@@ -264,6 +265,10 @@ async def set(
         if not self._is_open:
             await self._open()
         self._check_writable()
+        if not isinstance(value, Buffer):
+            raise TypeError(
+                f"FsspecStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
+            )
         path = _dereference_path(self.path, key)
         # write data
         if byte_range:

src/zarr/storage/_local.py

@@ -96,7 +96,7 @@ def __init__(self, root: Path | str, *, read_only: bool = False) -> None:
             root = Path(root)
         if not isinstance(root, Path):
             raise TypeError(
-                f'"root" must be a string or Path instance. Got an object with type {type(root)} instead.'
+                f"'root' must be a string or Path instance. Got an instance of {type(root)} instead."
             )
         self.root = root
 
@@ -169,7 +169,9 @@ async def _set(self, key: str, value: Buffer, exclusive: bool = False) -> None:
         self._check_writable()
         assert isinstance(key, str)
         if not isinstance(value, Buffer):
-            raise TypeError("LocalStore.set(): `value` must a Buffer instance")
+            raise TypeError(
+                f"LocalStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
+            )
         path = self.root / key
         await asyncio.to_thread(_put, path, value, start=None, exclusive=exclusive)
 

src/zarr/storage/_logging.py

@@ -2,10 +2,11 @@
 
 import inspect
 import logging
+import sys
 import time
 from collections import defaultdict
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Self, TypeVar
 
 from zarr.abc.store import Store
 from zarr.storage._wrapper import WrapperStore
@@ -18,8 +19,10 @@
 
     counter: defaultdict[str, int]
 
+T_Store = TypeVar("T_Store", bound=Store)
 
-class LoggingStore(WrapperStore[Store]):
+
+class LoggingStore(WrapperStore[T_Store]):
     """
     Store wrapper that logs all calls to the wrapped store.
 
@@ -42,7 +45,7 @@ class LoggingStore(WrapperStore[Store]):
 
     def __init__(
         self,
-        store: Store,
+        store: T_Store,
         log_level: str = "DEBUG",
         log_handler: logging.Handler | None = None,
     ) -> None:
@@ -67,7 +70,7 @@ def _configure_logger(
 
     def _default_handler(self) -> logging.Handler:
         """Define a default log handler"""
-        handler = logging.StreamHandler()
+        handler = logging.StreamHandler(stream=sys.stdout)
         handler.setLevel(self.log_level)
         handler.setFormatter(
             logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
@@ -94,6 +97,14 @@ def log(self, hint: Any = "") -> Generator[None, None, None]:
             end_time = time.time()
             self.logger.info("Finished %s [%.2f s]", op, end_time - start_time)
 
+    @classmethod
+    async def open(cls: type[Self], store_cls: type[T_Store], *args: Any, **kwargs: Any) -> Self:
+        log_level = kwargs.pop("log_level", "DEBUG")
+        log_handler = kwargs.pop("log_handler", None)
+        store = store_cls(*args, **kwargs)
+        await store._open()
+        return cls(store=store, log_level=log_level, log_handler=log_handler)
+
     @property
     def supports_writes(self) -> bool:
         with self.log():
@@ -126,8 +137,7 @@ def _is_open(self) -> bool:
 
     @_is_open.setter
     def _is_open(self, value: bool) -> None:
-        with self.log(value):
-            self._store._is_open = value
+        raise NotImplementedError("LoggingStore must be opened via the `_open` method")
 
     async def _open(self) -> None:
         with self.log():
@@ -151,11 +161,11 @@ def __str__(self) -> str:
         return f"logging-{self._store}"
 
     def __repr__(self) -> str:
-        return f"LoggingStore({repr(self._store)!r})"
+        return f"LoggingStore({self._store.__class__.__name__}, '{self._store}')"
 
     def __eq__(self, other: object) -> bool:
         with self.log(other):
-            return self._store == other
+            return type(self) is type(other) and self._store.__eq__(other._store)  # type: ignore[attr-defined]
 
     async def get(
         self,

src/zarr/storage/_memory.py

@@ -111,7 +111,9 @@ async def set(self, key: str, value: Buffer, byte_range: tuple[int, int] | None
         await self._ensure_open()
         assert isinstance(key, str)
         if not isinstance(value, Buffer):
-            raise TypeError(f"Expected Buffer. Got {type(value)}.")
+            raise TypeError(
+                f"MemoryStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
+            )
 
         if byte_range is not None:
             buf = self._store_dict[key]
@@ -231,8 +233,9 @@ async def set(self, key: str, value: Buffer, byte_range: tuple[int, int] | None
         self._check_writable()
         assert isinstance(key, str)
         if not isinstance(value, Buffer):
-            raise TypeError(f"Expected Buffer. Got {type(value)}.")
-
+            raise TypeError(
+                f"GpuMemoryStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
+            )
         # Convert to gpu.Buffer
         gpu_value = value if isinstance(value, gpu.Buffer) else gpu.Buffer.from_buffer(value)
         await super().set(key, gpu_value, byte_range=byte_range)

src/zarr/storage/_wrapper.py

@@ -56,6 +56,14 @@ async def _ensure_open(self) -> None:
     async def is_empty(self, prefix: str) -> bool:
         return await self._store.is_empty(prefix)
 
+    @property
+    def _is_open(self) -> bool:
+        return self._store._is_open
+
+    @_is_open.setter
+    def _is_open(self, value: bool) -> None:
+        raise NotImplementedError("WrapperStore must be opened via the `_open` method")
+
     async def clear(self) -> None:
         return await self._store.clear()
 
@@ -67,7 +75,13 @@ def _check_writable(self) -> None:
         return self._store._check_writable()
 
     def __eq__(self, value: object) -> bool:
-        return type(self) is type(value) and self._store.__eq__(value)
+        return type(self) is type(value) and self._store.__eq__(value._store)  # type: ignore[attr-defined]
+
+    def __str__(self) -> str:
+        return f"wrapping-{self._store}"
+
+    def __repr__(self) -> str:
+        return f"WrapperStore({self._store.__class__.__name__}, '{self._store}')"
 
     async def get(
         self, key: str, prototype: BufferPrototype, byte_range: ByteRequest | None = None

src/zarr/storage/_zip.py

@@ -107,11 +107,14 @@ def _sync_open(self) -> None:
     async def _open(self) -> None:
         self._sync_open()
 
-    def __getstate__(self) -> tuple[Path, ZipStoreAccessModeLiteral, int, bool]:
-        return self.path, self._zmode, self.compression, self.allowZip64
-
-    def __setstate__(self, state: Any) -> None:
-        self.path, self._zmode, self.compression, self.allowZip64 = state
+    def __getstate__(self) -> dict[str, Any]:
+        state = self.__dict__
+        for attr in ["_zf", "_lock"]:
+            state.pop(attr, None)
+        return state
+
+    def __setstate__(self, state: dict[str, Any]) -> None:
+        self.__dict__ = state
         self._is_open = False
         self._sync_open()
 
@@ -146,6 +149,8 @@ def _get(
         prototype: BufferPrototype,
         byte_range: ByteRequest | None = None,
     ) -> Buffer | None:
+        if not self._is_open:
+            self._sync_open()
         # docstring inherited
         try:
             with self._zf.open(key) as f:  # will raise KeyError
@@ -190,6 +195,8 @@ async def get_partial_values(
         return out
 
     def _set(self, key: str, value: Buffer) -> None:
+        if not self._is_open:
+            self._sync_open()
         # generally, this should be called inside a lock
         keyinfo = zipfile.ZipInfo(filename=key, date_time=time.localtime(time.time())[:6])
         keyinfo.compress_type = self.compression
@@ -203,9 +210,13 @@ def _set(self, key: str, value: Buffer) -> None:
     async def set(self, key: str, value: Buffer) -> None:
         # docstring inherited
         self._check_writable()
+        if not self._is_open:
+            self._sync_open()
         assert isinstance(key, str)
         if not isinstance(value, Buffer):
-            raise TypeError("ZipStore.set(): `value` must a Buffer instance")
+            raise TypeError(
+                f"ZipStore.set(): `value` must be a Buffer instance. Got an instance of {type(value)} instead."
+            )
         with self._lock:
             self._set(key, value)