Repository: zarr-developers/zarr-python

Issue:
Default to a RemoteStore for (some) string `StoreLike`
### Zarr version

v3

### Numcodecs version

n/a

### Python Version

n/a

### Operating System

n/a

### Installation

n/a

### Description

Zarr v2 users (either direct, or indirect through libraries like xarray) probably relied on the ability to pass an fsspec URI to `zarr.open` and have it work with a remote file system. Is there any tolerance for trying to interpret some string `store` arguments as an argument to `RemoteStore` rather than `LocalFile`? For example:

```python
zarr.open_group("s3://hrrrzarr")
```

would be interpreted as

```python
zarr.open_group(zarr.store.RemoteStore("s3://hrrrzarr"))
```


We'd also need to expose `storage_options` in all the API endpoints that create a `Store` / `StorePath` from user input, to allow customizing the filesystem that's created.

### Steps to reproduce

```python
import zarr

g = zarr.open_group(store="s3://hrrrzarr", storage_options={"anon": False})
```

### Additional output

_No response_

Discussion:

Patch:
diff --git a/src/zarr/api/asynchronous.py b/src/zarr/api/asynchronous.py
index 38576d73..58278172 100644
--- a/src/zarr/api/asynchronous.py
+++ b/src/zarr/api/asynchronous.py
@@ -194,6 +194,7 @@ async def open(
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     path: str | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: Any,  # TODO: type kwargs as valid args to open_array
 ) -> AsyncArray | AsyncGroup:
     """Convenience function to open a group or array using file-mode-like semantics.
@@ -211,6 +212,9 @@ async def open(
         The zarr format to use when saving.
     path : str or None, optional
         The path within the store to open.
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
     **kwargs
         Additional parameters are passed through to :func:`zarr.creation.open_array` or
         :func:`zarr.hierarchy.open_group`.
@@ -221,7 +225,7 @@ async def open(
         Return type depends on what exists in the given store.
     """
     zarr_format = _handle_zarr_version_or_format(zarr_version=zarr_version, zarr_format=zarr_format)
-    store_path = await make_store_path(store, mode=mode)
+    store_path = await make_store_path(store, mode=mode, storage_options=storage_options)
 
     if path is not None:
         store_path = store_path / path
@@ -276,6 +280,7 @@ async def save_array(
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     path: str | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: Any,  # TODO: type kwargs as valid args to create
 ) -> None:
     """Convenience function to save a NumPy array to the local file system, following a
@@ -291,6 +296,9 @@ async def save_array(
         The zarr format to use when saving.
     path : str or None, optional
         The path within the store where the array will be saved.
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
     kwargs
         Passed through to :func:`create`, e.g., compressor.
     """
@@ -299,7 +307,7 @@ async def save_array(
         or _default_zarr_version()
     )
 
-    store_path = await make_store_path(store, mode="w")
+    store_path = await make_store_path(store, mode="w", storage_options=storage_options)
     if path is not None:
         store_path = store_path / path
     new = await AsyncArray.create(
@@ -319,6 +327,7 @@ async def save_group(
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     path: str | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: NDArrayLike,
 ) -> None:
     """Convenience function to save several NumPy arrays to the local file system, following a
@@ -334,11 +343,17 @@ async def save_group(
         The zarr format to use when saving.
     path : str or None, optional
         Path within the store where the group will be saved.
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
     kwargs
         NumPy arrays with data to save.
     """
     zarr_format = (
-        _handle_zarr_version_or_format(zarr_version=zarr_version, zarr_format=zarr_format)
+        _handle_zarr_version_or_format(
+            zarr_version=zarr_version,
+            zarr_format=zarr_format,
+        )
         or _default_zarr_version()
     )
 
@@ -346,10 +361,22 @@ async def save_group(
         raise ValueError("at least one array must be provided")
     aws = []
     for i, arr in enumerate(args):
-        aws.append(save_array(store, arr, zarr_format=zarr_format, path=f"{path}/arr_{i}"))
+        aws.append(
+            save_array(
+                store,
+                arr,
+                zarr_format=zarr_format,
+                path=f"{path}/arr_{i}",
+                storage_options=storage_options,
+            )
+        )
     for k, arr in kwargs.items():
         _path = f"{path}/{k}" if path is not None else k
-        aws.append(save_array(store, arr, zarr_format=zarr_format, path=_path))
+        aws.append(
+            save_array(
+                store, arr, zarr_format=zarr_format, path=_path, storage_options=storage_options
+            )
+        )
     await asyncio.gather(*aws)
 
 
@@ -418,6 +445,7 @@ async def group(
     zarr_format: ZarrFormat | None = None,
     meta_array: Any | None = None,  # not used
     attributes: dict[str, JSON] | None = None,
+    storage_options: dict[str, Any] | None = None,
 ) -> AsyncGroup:
     """Create a group.
 
@@ -444,6 +472,9 @@ async def group(
         to users. Use `numpy.empty(())` by default.
     zarr_format : {2, 3, None}, optional
         The zarr format to use when saving.
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
 
     Returns
     -------
@@ -453,7 +484,7 @@ async def group(
 
     zarr_format = _handle_zarr_version_or_format(zarr_version=zarr_version, zarr_format=zarr_format)
 
-    store_path = await make_store_path(store)
+    store_path = await make_store_path(store, storage_options=storage_options)
     if path is not None:
         store_path = store_path / path
 
@@ -488,7 +519,7 @@ async def open_group(
     synchronizer: Any = None,  # not used
     path: str | None = None,
     chunk_store: StoreLike | None = None,  # not used
-    storage_options: dict[str, Any] | None = None,  # not used
+    storage_options: dict[str, Any] | None = None,
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     meta_array: Any | None = None,  # not used
@@ -548,10 +579,8 @@ async def open_group(
         warnings.warn("meta_array is not yet implemented", RuntimeWarning, stacklevel=2)
     if chunk_store is not None:
         warnings.warn("chunk_store is not yet implemented", RuntimeWarning, stacklevel=2)
-    if storage_options is not None:
-        warnings.warn("storage_options is not yet implemented", RuntimeWarning, stacklevel=2)
 
-    store_path = await make_store_path(store, mode=mode)
+    store_path = await make_store_path(store, mode=mode, storage_options=storage_options)
     if path is not None:
         store_path = store_path / path
 
@@ -603,6 +632,7 @@ async def create(
     ) = None,
     codecs: Iterable[Codec | dict[str, JSON]] | None = None,
     dimension_names: Iterable[str] | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: Any,
 ) -> AsyncArray:
     """Create an array.
@@ -674,6 +704,9 @@ async def create(
         to users. Use `numpy.empty(())` by default.
 
         .. versionadded:: 2.13
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
 
     Returns
     -------
@@ -725,7 +758,7 @@ async def create(
         warnings.warn("meta_array is not yet implemented", RuntimeWarning, stacklevel=2)
 
     mode = kwargs.pop("mode", cast(AccessModeLiteral, "r" if read_only else "w"))
-    store_path = await make_store_path(store, mode=mode)
+    store_path = await make_store_path(store, mode=mode, storage_options=storage_options)
     if path is not None:
         store_path = store_path / path
 
@@ -875,6 +908,7 @@ async def open_array(
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     path: PathLike | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: Any,  # TODO: type kwargs as valid args to save
 ) -> AsyncArray:
     """Open an array using file-mode-like semantics.
@@ -887,6 +921,9 @@ async def open_array(
         The zarr format to use when saving.
     path : string, optional
         Path in store to array.
+    storage_options : dict
+        If using an fsspec URL to create the store, these will be passed to
+        the backend implementation. Ignored otherwise.
     **kwargs
         Any keyword arguments to pass to the array constructor.
 
@@ -896,7 +933,7 @@ async def open_array(
         The opened array.
     """
 
-    store_path = await make_store_path(store)
+    store_path = await make_store_path(store, storage_options=storage_options)
     if path is not None:
         store_path = store_path / path
 
diff --git a/src/zarr/api/synchronous.py b/src/zarr/api/synchronous.py
index 93a33b8d..129f9019 100644
--- a/src/zarr/api/synchronous.py
+++ b/src/zarr/api/synchronous.py
@@ -134,6 +134,7 @@ def save_group(
     zarr_version: ZarrFormat | None = None,  # deprecated
     zarr_format: ZarrFormat | None = None,
     path: str | None = None,
+    storage_options: dict[str, Any] | None = None,
     **kwargs: NDArrayLike,
 ) -> None:
     return sync(
@@ -143,6 +144,7 @@ def save_group(
             zarr_version=zarr_version,
             zarr_format=zarr_format,
             path=path,
+            storage_options=storage_options,
             **kwargs,
         )
     )
diff --git a/src/zarr/store/common.py b/src/zarr/store/common.py
index 8028c9af..196479dd 100644
--- a/src/zarr/store/common.py
+++ b/src/zarr/store/common.py
@@ -11,6 +11,8 @@ from zarr.errors import ContainsArrayAndGroupError, ContainsArrayError, Contains
 from zarr.store.local import LocalStore
 from zarr.store.memory import MemoryStore
 
+# from zarr.store.remote import RemoteStore
+
 if TYPE_CHECKING:
     from zarr.core.buffer import BufferPrototype
     from zarr.core.common import AccessModeLiteral
@@ -75,30 +77,69 @@ StoreLike = Store | StorePath | Path | str | dict[str, Buffer]
 
 
 async def make_store_path(
-    store_like: StoreLike | None, *, mode: AccessModeLiteral | None = None
+    store_like: StoreLike | None,
+    *,
+    mode: AccessModeLiteral | None = None,
+    storage_options: dict[str, Any] | None = None,
 ) -> StorePath:
+    from zarr.store.remote import RemoteStore  # circular import
+
+    used_storage_options = False
+
     if isinstance(store_like, StorePath):
         if mode is not None:
             assert AccessMode.from_literal(mode) == store_like.store.mode
-        return store_like
+        result = store_like
     elif isinstance(store_like, Store):
         if mode is not None:
             assert AccessMode.from_literal(mode) == store_like.mode
         await store_like._ensure_open()
-        return StorePath(store_like)
+        result = StorePath(store_like)
     elif store_like is None:
         if mode is None:
             mode = "w"  # exception to the default mode = 'r'
-        return StorePath(await MemoryStore.open(mode=mode))
+        result = StorePath(await MemoryStore.open(mode=mode))
     elif isinstance(store_like, Path):
-        return StorePath(await LocalStore.open(root=store_like, mode=mode or "r"))
+        result = StorePath(await LocalStore.open(root=store_like, mode=mode or "r"))
     elif isinstance(store_like, str):
-        return StorePath(await LocalStore.open(root=Path(store_like), mode=mode or "r"))
+        storage_options = storage_options or {}
+
+        if _is_fsspec_uri(store_like):
+            used_storage_options = True
+            result = StorePath(
+                RemoteStore.from_url(store_like, storage_options=storage_options, mode=mode or "r")
+            )
+        else:
+            result = StorePath(await LocalStore.open(root=Path(store_like), mode=mode or "r"))
     elif isinstance(store_like, dict):
         # We deliberate only consider dict[str, Buffer] here, and not arbitrary mutable mappings.
         # By only allowing dictionaries, which are in-memory, we know that MemoryStore appropriate.
-        return StorePath(await MemoryStore.open(store_dict=store_like, mode=mode))
-    raise TypeError
+        result = StorePath(await MemoryStore.open(store_dict=store_like, mode=mode))
+    else:
+        msg = f"Unsupported type for store_like: '{type(store_like).__name__}'"  # type: ignore[unreachable]
+        raise TypeError(msg)
+
+    if storage_options and not used_storage_options:
+        msg = "'storage_options' was provided but unused. 'storage_options' is only used for fsspec filesystem stores."
+        raise TypeError(msg)
+
+    return result
+
+
+def _is_fsspec_uri(uri: str) -> bool:
+    """
+    Check if a URI looks like a non-local fsspec URI.
+
+    Examples
+    --------
+    >>> _is_fsspec_uri("s3://bucket")
+    True
+    >>> _is_fsspec_uri("my-directory")
+    False
+    >>> _is_fsspec_uri("local://my-directory")
+    False
+    """
+    return "://" in uri or "::" in uri and "local://" not in uri
 
 
 async def ensure_no_existing_node(store_path: StorePath, zarr_format: ZarrFormat) -> None:
diff --git a/src/zarr/store/memory.py b/src/zarr/store/memory.py
index e3044197..7baa6aee 100644
--- a/src/zarr/store/memory.py
+++ b/src/zarr/store/memory.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncGenerator, MutableMapping
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from zarr.abc.store import Store
 from zarr.core.buffer import Buffer, gpu
@@ -55,12 +55,6 @@ class MemoryStore(Store):
             and self.mode == other.mode
         )
 
-    def __setstate__(self, state: Any) -> None:
-        raise NotImplementedError(f"{type(self)} cannot be pickled")
-
-    def __getstate__(self) -> None:
-        raise NotImplementedError(f"{type(self)} cannot be pickled")
-
     async def get(
         self,
         key: str,
diff --git a/src/zarr/store/remote.py b/src/zarr/store/remote.py
index 084ef986..ecb46a31 100644
--- a/src/zarr/store/remote.py
+++ b/src/zarr/store/remote.py
@@ -11,12 +11,18 @@ if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
     from fsspec.asyn import AsyncFileSystem
-    from upath import UPath
 
     from zarr.core.buffer import Buffer, BufferPrototype
     from zarr.core.common import AccessModeLiteral, BytesLike
 
 
+ALLOWED_EXCEPTIONS: tuple[type[Exception], ...] = (
+    FileNotFoundError,
+    IsADirectoryError,
+    NotADirectoryError,
+)
+
+
 class RemoteStore(Store):
     # based on FSSpec
     supports_writes: bool = True
@@ -24,21 +30,15 @@ class RemoteStore(Store):
     supports_partial_writes: bool = False
     supports_listing: bool = True
 
-    _fs: AsyncFileSystem
-    _url: str
-    path: str
+    fs: AsyncFileSystem
     allowed_exceptions: tuple[type[Exception], ...]
 
     def __init__(
         self,
-        url: UPath | str,
+        fs: AsyncFileSystem,
         mode: AccessModeLiteral = "r",
-        allowed_exceptions: tuple[type[Exception], ...] = (
-            FileNotFoundError,
-            IsADirectoryError,
-            NotADirectoryError,
-        ),
-        **storage_options: Any,
+        path: str = "/",
+        allowed_exceptions: tuple[type[Exception], ...] = ALLOWED_EXCEPTIONS,
     ):
         """
         Parameters
@@ -51,54 +51,58 @@ class RemoteStore(Store):
             this must not be used.
         """
         super().__init__(mode=mode)
-        self._storage_options = storage_options
-        if isinstance(url, str):
-            self._url = url.rstrip("/")
-            self._fs, _path = fsspec.url_to_fs(url, **storage_options)
-            self.path = _path.rstrip("/")
-        elif hasattr(url, "protocol") and hasattr(url, "fs"):
-            # is UPath-like - but without importing
-            if storage_options:
-                raise ValueError(
-                    "If constructed with a UPath object, no additional "
-                    "storage_options are allowed"
-                )
-            # n.b. UPath returns the url and path attributes with a trailing /, at least for s3
-            # that trailing / must be removed to compose with the store interface
-            self._url = str(url).rstrip("/")
-            self.path = url.path.rstrip("/")
-            self._fs = url.fs
-        else:
-            raise ValueError(f"URL not understood, {url}")
+        self.fs = fs
+        self.path = path
         self.allowed_exceptions = allowed_exceptions
-        # test instantiate file system
-        if not self._fs.async_impl:
-            raise TypeError("FileSystem needs to support async operations")
+
+        if not self.fs.async_impl:
+            raise TypeError("Filesystem needs to support async operations.")
+
+    @classmethod
+    def from_upath(
+        cls,
+        upath: Any,
+        mode: AccessModeLiteral = "r",
+        allowed_exceptions: tuple[type[Exception], ...] = ALLOWED_EXCEPTIONS,
+    ) -> RemoteStore:
+        return cls(
+            fs=upath.fs,
+            path=upath.path.rstrip("/"),
+            mode=mode,
+            allowed_exceptions=allowed_exceptions,
+        )
+
+    @classmethod
+    def from_url(
+        cls,
+        url: str,
+        storage_options: dict[str, Any] | None = None,
+        mode: AccessModeLiteral = "r",
+        allowed_exceptions: tuple[type[Exception], ...] = ALLOWED_EXCEPTIONS,
+    ) -> RemoteStore:
+        fs, path = fsspec.url_to_fs(url, **storage_options)
+        return cls(fs=fs, path=path, mode=mode, allowed_exceptions=allowed_exceptions)
 
     async def clear(self) -> None:
         try:
-            for subpath in await self._fs._find(self.path, withdirs=True):
+            for subpath in await self.fs._find(self.path, withdirs=True):
                 if subpath != self.path:
-                    await self._fs._rm(subpath, recursive=True)
+                    await self.fs._rm(subpath, recursive=True)
         except FileNotFoundError:
             pass
 
     async def empty(self) -> bool:
-        return not await self._fs._find(self.path, withdirs=True)
-
-    def __str__(self) -> str:
-        return f"{self._url}"
+        return not await self.fs._find(self.path, withdirs=True)
 
     def __repr__(self) -> str:
-        return f"<RemoteStore({type(self._fs).__name__}, {self.path})>"
+        return f"<RemoteStore({type(self.fs).__name__}, {self.path})>"
 
     def __eq__(self, other: object) -> bool:
         return (
             isinstance(other, type(self))
             and self.path == other.path
             and self.mode == other.mode
-            and self._url == other._url
-            # and self._storage_options == other._storage_options  # FIXME: this isn't working for some reason
+            and self.fs == other.fs
         )
 
     async def get(
@@ -123,9 +127,9 @@ class RemoteStore(Store):
                     end = None
             value = prototype.buffer.from_bytes(
                 await (
-                    self._fs._cat_file(path, start=byte_range[0], end=end)
+                    self.fs._cat_file(path, start=byte_range[0], end=end)
                     if byte_range
-                    else self._fs._cat_file(path)
+                    else self.fs._cat_file(path)
                 )
             )
 
@@ -152,13 +156,13 @@ class RemoteStore(Store):
         # write data
         if byte_range:
             raise NotImplementedError
-        await self._fs._pipe_file(path, value.to_bytes())
+        await self.fs._pipe_file(path, value.to_bytes())
 
     async def delete(self, key: str) -> None:
         self._check_writable()
         path = _dereference_path(self.path, key)
         try:
-            await self._fs._rm(path)
+            await self.fs._rm(path)
         except FileNotFoundError:
             pass
         except self.allowed_exceptions:
@@ -166,7 +170,7 @@ class RemoteStore(Store):
 
     async def exists(self, key: str) -> bool:
         path = _dereference_path(self.path, key)
-        exists: bool = await self._fs._exists(path)
+        exists: bool = await self.fs._exists(path)
         return exists
 
     async def get_partial_values(
@@ -189,7 +193,7 @@ class RemoteStore(Store):
         else:
             return []
         # TODO: expectations for exceptions or missing keys?
-        res = await self._fs._cat_ranges(list(paths), starts, stops, on_error="return")
+        res = await self.fs._cat_ranges(list(paths), starts, stops, on_error="return")
         # the following is an s3-specific condition we probably don't want to leak
         res = [b"" if (isinstance(r, OSError) and "not satisfiable" in str(r)) else r for r in res]
         for r in res:
@@ -202,14 +206,14 @@ class RemoteStore(Store):
         raise NotImplementedError
 
     async def list(self) -> AsyncGenerator[str, None]:
-        allfiles = await self._fs._find(self.path, detail=False, withdirs=False)
+        allfiles = await self.fs._find(self.path, detail=False, withdirs=False)
         for onefile in (a.replace(self.path + "/", "") for a in allfiles):
             yield onefile
 
     async def list_dir(self, prefix: str) -> AsyncGenerator[str, None]:
         prefix = f"{self.path}/{prefix.rstrip('/')}"
         try:
-            allfiles = await self._fs._ls(prefix, detail=False)
+            allfiles = await self.fs._ls(prefix, detail=False)
         except FileNotFoundError:
             return
         for onefile in (a.replace(prefix + "/", "") for a in allfiles):
@@ -230,5 +234,5 @@ class RemoteStore(Store):
         """
 
         find_str = "/".join([self.path, prefix])
-        for onefile in await self._fs._find(find_str, detail=False, maxdepth=None, withdirs=False):
+        for onefile in await self.fs._find(find_str, detail=False, maxdepth=None, withdirs=False):
             yield onefile.removeprefix(find_str)


Test patch:
diff --git a/tests/v3/test_store/test_core.py b/tests/v3/test_store/test_core.py
index c65d91f9..f4014911 100644
--- a/tests/v3/test_store/test_core.py
+++ b/tests/v3/test_store/test_core.py
@@ -1,10 +1,12 @@
+import tempfile
 from pathlib import Path
 
 import pytest
 
-from zarr.store.common import make_store_path
+from zarr.store.common import StoreLike, StorePath, make_store_path
 from zarr.store.local import LocalStore
 from zarr.store.memory import MemoryStore
+from zarr.store.remote import RemoteStore
 
 
 async def test_make_store_path(tmpdir: str) -> None:
@@ -34,3 +36,32 @@ async def test_make_store_path(tmpdir: str) -> None:
 
     with pytest.raises(TypeError):
         await make_store_path(1)  # type: ignore[arg-type]
+
+
+async def test_make_store_path_fsspec(monkeypatch) -> None:
+    import fsspec.implementations.memory
+
+    monkeypatch.setattr(fsspec.implementations.memory.MemoryFileSystem, "async_impl", True)
+    store_path = await make_store_path("memory://")
+    assert isinstance(store_path.store, RemoteStore)
+
+
+@pytest.mark.parametrize(
+    "store_like",
+    [
+        None,
+        str(tempfile.TemporaryDirectory()),
+        Path(tempfile.TemporaryDirectory().name),
+        StorePath(store=MemoryStore(store_dict={}, mode="w"), path="/"),
+        MemoryStore(store_dict={}, mode="w"),
+        {},
+    ],
+)
+async def test_make_store_path_storage_options_raises(store_like: StoreLike) -> None:
+    with pytest.raises(TypeError, match="storage_options"):
+        await make_store_path(store_like, storage_options={"foo": "bar"}, mode="w")
+
+
+async def test_unsupported() -> None:
+    with pytest.raises(TypeError, match="Unsupported type for store_like: 'int'"):
+        await make_store_path(1)  # type: ignore[arg-type]
diff --git a/tests/v3/test_store/test_memory.py b/tests/v3/test_store/test_memory.py
index 470383df..2498cdc2 100644
--- a/tests/v3/test_store/test_memory.py
+++ b/tests/v3/test_store/test_memory.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import pickle
-
 import pytest
 
 from zarr.core.buffer import Buffer, cpu, gpu
@@ -48,16 +46,6 @@ class TestMemoryStore(StoreTests[MemoryStore, cpu.Buffer]):
     def test_list_prefix(self, store: MemoryStore) -> None:
         assert True
 
-    def test_serizalizable_store(self, store: MemoryStore) -> None:
-        with pytest.raises(NotImplementedError):
-            store.__getstate__()
-
-        with pytest.raises(NotImplementedError):
-            store.__setstate__({})
-
-        with pytest.raises(NotImplementedError):
-            pickle.dumps(store)
-
 
 @gpu_test
 class TestGpuMemoryStore(StoreTests[GpuMemoryStore, gpu.Buffer]):
@@ -92,13 +80,3 @@ class TestGpuMemoryStore(StoreTests[GpuMemoryStore, gpu.Buffer]):
 
     def test_list_prefix(self, store: GpuMemoryStore) -> None:
         assert True
-
-    def test_serizalizable_store(self, store: MemoryStore) -> None:
-        with pytest.raises(NotImplementedError):
-            store.__getstate__()
-
-        with pytest.raises(NotImplementedError):
-            store.__setstate__({})
-
-        with pytest.raises(NotImplementedError):
-            pickle.dumps(store)
diff --git a/tests/v3/test_store/test_remote.py b/tests/v3/test_store/test_remote.py
index e2c30701..495a5e5c 100644
--- a/tests/v3/test_store/test_remote.py
+++ b/tests/v3/test_store/test_remote.py
@@ -1,24 +1,27 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from collections.abc import Generator
-
-    import botocore.client
-
+import json
 import os
+from collections.abc import Generator
+from typing import TYPE_CHECKING
 
 import fsspec
 import pytest
 from botocore.session import Session
 from upath import UPath
 
+import zarr.api.asynchronous
 from zarr.core.buffer import Buffer, cpu, default_buffer_prototype
 from zarr.core.sync import _collect_aiterator, sync
 from zarr.store import RemoteStore
 from zarr.testing.store import StoreTests
 
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    import botocore.client
+
+
 s3fs = pytest.importorskip("s3fs")
 requests = pytest.importorskip("requests")
 moto_server = pytest.importorskip("moto.moto_server.threaded_moto_server")
@@ -90,8 +93,10 @@ async def alist(it):
 
 
 async def test_basic() -> None:
-    store = await RemoteStore.open(
-        f"s3://{test_bucket_name}", mode="w", endpoint_url=endpoint_url, anon=False
+    store = RemoteStore.from_url(
+        f"s3://{test_bucket_name}",
+        mode="w",
+        storage_options=dict(endpoint_url=endpoint_url, anon=False),
     )
     assert await _collect_aiterator(store.list()) == ()
     assert not await store.exists("foo")
@@ -109,51 +114,33 @@ class TestRemoteStoreS3(StoreTests[RemoteStore, cpu.Buffer]):
     store_cls = RemoteStore
     buffer_cls = cpu.Buffer
 
-    @pytest.fixture(scope="function", params=("use_upath", "use_str"))
-    def store_kwargs(self, request: pytest.FixtureRequest) -> dict[str, str | bool | UPath]:  # type: ignore
-        url = f"s3://{test_bucket_name}"
-        anon = False
-        mode = "r+"
-        if request.param == "use_upath":
-            return {"url": UPath(url, endpoint_url=endpoint_url, anon=anon), "mode": mode}
-        elif request.param == "use_str":
-            return {"url": url, "mode": mode, "anon": anon, "endpoint_url": endpoint_url}
-
-        raise AssertionError
+    @pytest.fixture(scope="function")
+    def store_kwargs(self, request) -> dict[str, str | bool]:
+        fs, path = fsspec.url_to_fs(
+            f"s3://{test_bucket_name}", endpoint_url=endpoint_url, anon=False
+        )
+        return {"fs": fs, "path": path, "mode": "r+"}
 
     @pytest.fixture(scope="function")
-    async def store(self, store_kwargs: dict[str, str | bool | UPath]) -> RemoteStore:
-        url: str | UPath = store_kwargs["url"]
-        mode = store_kwargs["mode"]
-        if isinstance(url, UPath):
-            out = self.store_cls(url=url, mode=mode)
-        else:
-            endpoint_url = store_kwargs["endpoint_url"]
-            out = self.store_cls(url=url, asynchronous=True, mode=mode, endpoint_url=endpoint_url)
-        return out
+    def store(self, store_kwargs: dict[str, str | bool]) -> RemoteStore:
+        return self.store_cls(**store_kwargs)
 
     def get(self, store: RemoteStore, key: str) -> Buffer:
         #  make a new, synchronous instance of the filesystem because this test is run in sync code
-        fs, _ = fsspec.url_to_fs(
-            url=store._url,
-            asynchronous=False,
-            anon=store._fs.anon,
-            endpoint_url=store._fs.endpoint_url,
+        new_fs = fsspec.filesystem(
+            "s3", endpoint_url=store.fs.endpoint_url, anon=store.fs.anon, asynchronous=False
         )
-        return self.buffer_cls.from_bytes(fs.cat(f"{store.path}/{key}"))
+        return self.buffer_cls.from_bytes(new_fs.cat(f"{store.path}/{key}"))
 
     def set(self, store: RemoteStore, key: str, value: Buffer) -> None:
         #  make a new, synchronous instance of the filesystem because this test is run in sync code
-        fs, _ = fsspec.url_to_fs(
-            url=store._url,
-            asynchronous=False,
-            anon=store._fs.anon,
-            endpoint_url=store._fs.endpoint_url,
+        new_fs = fsspec.filesystem(
+            "s3", endpoint_url=store.fs.endpoint_url, anon=store.fs.anon, asynchronous=False
         )
-        fs.write_bytes(f"{store.path}/{key}", value.to_bytes())
+        new_fs.write_bytes(f"{store.path}/{key}", value.to_bytes())
 
     def test_store_repr(self, store: RemoteStore) -> None:
-        assert str(store) == f"s3://{test_bucket_name}"
+        assert str(store) == "<RemoteStore(S3FileSystem, test)>"
 
     def test_store_supports_writes(self, store: RemoteStore) -> None:
         assert True
@@ -164,3 +151,47 @@ class TestRemoteStoreS3(StoreTests[RemoteStore, cpu.Buffer]):
 
     def test_store_supports_listing(self, store: RemoteStore) -> None:
         assert True
+
+    async def test_remote_store_from_uri(
+        self, store: RemoteStore, store_kwargs: dict[str, str | bool]
+    ):
+        storage_options = {
+            "endpoint_url": endpoint_url,
+            "anon": False,
+        }
+
+        meta = {"attributes": {"key": "value"}, "zarr_format": 3, "node_type": "group"}
+
+        await store.set(
+            "zarr.json",
+            self.buffer_cls.from_bytes(json.dumps(meta).encode()),
+        )
+        group = await zarr.api.asynchronous.open_group(
+            store=f"s3://{test_bucket_name}", storage_options=storage_options
+        )
+        assert dict(group.attrs) == {"key": "value"}
+
+        meta["attributes"]["key"] = "value-2"
+        await store.set(
+            "directory-2/zarr.json",
+            self.buffer_cls.from_bytes(json.dumps(meta).encode()),
+        )
+        group = await zarr.api.asynchronous.open_group(
+            store=f"s3://{test_bucket_name}/directory-2", storage_options=storage_options
+        )
+        assert dict(group.attrs) == {"key": "value-2"}
+
+        meta["attributes"]["key"] = "value-3"
+        await store.set(
+            "directory-3/zarr.json",
+            self.buffer_cls.from_bytes(json.dumps(meta).encode()),
+        )
+        group = await zarr.api.asynchronous.open_group(
+            store=f"s3://{test_bucket_name}", path="directory-3", storage_options=storage_options
+        )
+        assert dict(group.attrs) == {"key": "value-3"}
+
+    def test_from_upath(self) -> None:
+        path = UPath(f"s3://{test_bucket_name}", endpoint_url=endpoint_url, anon=False)
+        result = RemoteStore.from_upath(path)
+        assert result.fs.endpoint_url == endpoint_url