Fix zarr v3 compression: use native codecs and bump min deps

Use "compressors" (plural) with native zarr v3 codec objects (ZstdCodec,
BloscCodec) in storage_options, which ome-zarr-py >= 0.16.0 + dask >= 2026.3.0
correctly forwards to zarr_array_kwargs. Bump minimum deps accordingly and fix
test assertions for lz4 (.clevel) vs zstd (.level) and the s0 array path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: LucaMarconato

pyproject.toml

@@ -26,16 +26,16 @@ dependencies = [
     "annsel>=0.1.2",
     "click",
     "dask-image",
-    "dask>=2025.12.0,<2026.1.2",
-    "distributed<2026.1.2",
+    "dask>=2026.3.0",
+    "distributed>=2026.3.0",
     "datashader",
     "fsspec[s3,http]",
     "geopandas>=0.14",
     "multiscale_spatial_image==2.0.3",
     "networkx",
     "numba>=0.55.0",
     "numpy",
-    "ome_zarr>=0.14.0",
+    "ome_zarr>=0.16.0",
     "pandas",
     "pooch",
     "pyarrow",

src/spatialdata/_io/io_raster.py

@@ -342,6 +342,21 @@ def _write_raster(
     group.attrs[ATTRS_KEY] = attrs
 
 
+def _build_v3_codec(
+    compression: Literal["lz4", "zstd"],
+    compression_level: int,
+) -> Any:
+    """Return the appropriate zarr v3 codec for the given compression type and level."""
+    if compression == "zstd":
+        from zarr.codecs import ZstdCodec
+
+        return ZstdCodec(level=compression_level)
+    # lz4: use the native zarr v3 BloscCodec
+    from zarr.codecs import BloscCodec
+
+    return BloscCodec(cname="lz4", clevel=compression_level)
+
+
 def _apply_compression(
     storage_options: JSONDict | list[JSONDict],
     compressor: dict[Literal["lz4", "zstd"], int] | None,
@@ -362,32 +377,47 @@ def _apply_compression(
     -------
     Updated storage options with compression settings
     """
-    # For zarr disk format v2, use numcodecs.Blosc
-    # For zarr disk format v3, use zarr.codecs.Blosc
-    from numcodecs import Blosc as BloscV2
-    from zarr.codecs import Blosc as BloscV3
-
     if not compressor:
         return storage_options
 
     ((compression, compression_level),) = compressor.items()
 
-    assert BloscV2.SHUFFLE == 1
-    blosc_v2 = BloscV2(cname=compression, clevel=compression_level, shuffle=1)
-    blosc_v3 = BloscV3(cname=compression, clevel=compression_level, shuffle=1)
+    if zarr_format == 2:
+        from numcodecs import Blosc as BloscV2
 
-    def _update_dict(d: dict[str, Any]) -> None:
-        if zarr_format == 2:
-            d["compressor"] = blosc_v2
-        elif zarr_format == 3:
-            d["zarr_array_kwargs"] = {"compressors": [blosc_v3]}
+        assert BloscV2.SHUFFLE == 1
+        codec_v2 = BloscV2(cname=compression, clevel=compression_level, shuffle=1)
 
-    if isinstance(storage_options, dict):
-        _update_dict(d=storage_options)
+        def _update_dict(d: dict[str, Any]) -> None:
+            d["compressor"] = codec_v2
+
+        if isinstance(storage_options, dict):
+            _update_dict(d=storage_options)
+        elif isinstance(storage_options, list):
+            for option in storage_options:
+                _update_dict(d=option)
+        elif storage_options is None:
+            return {"compressor": codec_v2}
+        else:
+            raise ValueError(f"storage_options must be a dict or list, not {type(storage_options)}")
     else:
-        assert isinstance(storage_options, list)
-        for option in storage_options:
-            _update_dict(d=option)
+        # zarr v3: use native codec objects via the "compressors" (plural) key.
+        # see  https://github.com/ome/ome-zarr-py/blob/v0.16.0/ome_zarr/writer.py#L754
+        # ome-zarr-py ≥ 0.16.0 with dask ≥ 2026.3.0 forwards this key to zarr_array_kwargs.
+        codec_v3 = _build_v3_codec(compression, compression_level)
+
+        def _update_dict_v3(d: dict[str, Any]) -> None:
+            d["compressors"] = [codec_v3]
+
+        if isinstance(storage_options, dict):
+            _update_dict_v3(d=storage_options)
+        elif isinstance(storage_options, list):
+            for option in storage_options:
+                _update_dict_v3(d=option)
+        elif storage_options is None:
+            return {"compressors": [codec_v3]}
+        else:
+            raise ValueError(f"storage_options must be a dict or list, not {type(storage_options)}")
 
     return storage_options
 

tests/io/test_readwrite.py

@@ -265,10 +265,9 @@ def test_compression_roundtrip(
         # sourcery skip: no-loop-in-tests
         for element in ["image2d", "image2d_multiscale", "labels2d", "labels2d_multiscale"]:
             element_type = "images" if element.startswith("image") else "labels"
-            arr = zarr.open_group(tmpdir / element_type, mode="r")[element]["0"]
+            arr = zarr.open_group(tmpdir / element_type, mode="r")[element]["s0"]
             compressor = arr.compressors[0]
 
-            # TODO: all these tests fail because the compression arguments are not passed to Dask
             if sdata_container_format.zarr_format == 2:
                 assert compressor.cname == "zstd"
                 assert compressor.clevel == 8
@@ -295,23 +294,23 @@ def test_write_element_compression(
         sdata["element"] = full_sdata[element[1]]
         sdata.write_element("element", compressor=compressor, sdata_formats=sdata_container_format)
 
-        arr = zarr.open_group(tmpdir / element[0], mode="r")["element"]["0"]
+        arr = zarr.open_group(tmpdir / element[0], mode="r")["element"]["s0"]
         compression = arr.compressors[0]
 
-        # TODO: all these tests fail because the compression arguments are not passed to Dask
         if sdata_container_format.zarr_format == 2:
             assert compression.cname == list(compressor.keys())[0]
             assert compression.clevel == list(compressor.values())[0]
         elif sdata_container_format.zarr_format == 3:
-            from zarr.codecs import ZstdCodec
+            from zarr.codecs import BloscCodec, ZstdCodec
 
             compressor_name = list(compressor.keys())[0]
+            compressor_level = list(compressor.values())[0]
             if compressor_name == "zstd":
                 assert isinstance(compression, ZstdCodec)
-            # TODO: fix
-            # elif compressor_name == 'lz4':
-            #     assert isinstance(compression, ???)
-            assert compression.level == list(compressor.values())[0]
+                assert compression.level == compressor_level
+            elif compressor_name == "lz4":
+                assert isinstance(compression, BloscCodec)
+                assert compression.clevel == compressor_level
 
     def test_incremental_io_list_of_elements(
         self,