Add with_read_only() convenience method to store

src/zarr/abc/store.py

@@ -83,6 +83,22 @@
         await store._open()
         return store
 
+    def with_read_only(self, read_only: bool = False) -> Store:
+        """
+        Return a new store of the same type pointing to the same location with the specified read_only state.
+        The returned Store is not automatically opened.
+
+        Parameters
+        ----------
+        read_only
+            If True, the store will be created in read-only mode. Defaults to False.
+
+        Returns
+        -------
+            A new store of the same type with the new read only attribute.
+        """
+        raise NotImplementedError("with_read_only is not implemented for this store type.")
+
     def __enter__(self) -> Self:
         """Enter a context manager that will close the store upon exiting."""
         return self

src/zarr/storage/_memory.py

@@ -54,6 +54,13 @@
             store_dict = {}
         self._store_dict = store_dict
 
+    def with_read_only(self, read_only: bool = False) -> MemoryStore:
+        # docstring inherited
+        return type(self)(
+            store_dict=self._store_dict,
+            read_only=read_only,
+        )
+
     async def clear(self) -> None:
         # docstring inherited
         self._store_dict.clear()

src/zarr/testing/store.py

@@ -149,6 +149,14 @@
         ):
             await store.delete("foo")
 
+    async def test_with_read_only_store(self, open_kwargs: dict[str, Any]) -> None:
+        kwargs = {**open_kwargs, "read_only": True}
+        store = await self.store_cls.open(**kwargs)
+        with pytest.raises(
+            NotImplementedError, match="with_read_only is not implemented for this store type."
+        ):
+            store.with_read_only(read_only=False)
+
     @pytest.mark.parametrize("key", ["c/0", "foo/c/0.0", "foo/0/0"])
     @pytest.mark.parametrize(
         ("data", "byte_range"),

tests/test_store/test_memory.py

@@ -78,6 +78,46 @@ async def test_deterministic_size(
         np.testing.assert_array_equal(a[:3], 1)
         np.testing.assert_array_equal(a[3:], 0)
 
+    async def test_with_read_only_store(self, open_kwargs: dict[str, Any]) -> None:
+        kwargs = {**open_kwargs, "read_only": True}
+        store = await self.store_cls.open(**kwargs)
+        assert store.read_only
+
+        # Test that you cannot write to a read-only store
+        with pytest.raises(
+            ValueError, match="store was opened in read-only mode and does not support writing"
+        ):
+            await store.set("foo", self.buffer_cls.from_bytes(b"bar"))
+
+        # Test that you can write to a copy that is not read-only
+        writer = store.with_read_only(read_only=False)
+        assert not writer._is_open
+        assert not writer.read_only
+        await writer.set("foo", self.buffer_cls.from_bytes(b"bar"))
+        await writer.delete("foo")
+
+        # Test that you cannot write to the original store
+        assert store.read_only
+        with pytest.raises(
+            ValueError, match="store was opened in read-only mode and does not support writing"
+        ):
+            await store.set("foo", self.buffer_cls.from_bytes(b"bar"))
+        with pytest.raises(
+            ValueError, match="store was opened in read-only mode and does not support writing"
+        ):
+            await store.delete("foo")
+        # Test that you cannot write to a read-only store copy
+        reader = store.with_read_only(read_only=True)
+        assert reader.read_only
+        with pytest.raises(
+            ValueError, match="store was opened in read-only mode and does not support writing"
+        ):
+            await reader.set("foo", self.buffer_cls.from_bytes(b"bar"))
+        with pytest.raises(
+            ValueError, match="store was opened in read-only mode and does not support writing"
+        ):
+            await reader.delete("foo")
+
 
 # TODO: fix this warning
 @pytest.mark.filterwarnings("ignore:Unclosed client session:ResourceWarning")