Backport PR #3310 on branch 3.1.x (obstore delete_dir)

changes/3310.feature.rst

@@ -0,0 +1 @@
+Add obstore implementation of delete_dir.

src/zarr/storage/_obstore.py

@@ -13,6 +13,7 @@
     Store,
     SuffixByteRequest,
 )
+from zarr.core.common import concurrent_map
 from zarr.core.config import config
 
 if TYPE_CHECKING:
@@ -196,6 +197,18 @@ async def delete(self, key: str) -> None:
         with contextlib.suppress(FileNotFoundError):
             await obs.delete_async(self.store, key)
 
+    async def delete_dir(self, prefix: str) -> None:
+        # docstring inherited
+        import obstore as obs
+
+        self._check_writable()
+        if prefix != "" and not prefix.endswith("/"):
+            prefix += "/"
+
+        metas = await obs.list(self.store, prefix).collect_async()
+        keys = [(m["path"],) for m in metas]
+        await concurrent_map(keys, self.delete, limit=config.get("async.concurrency"))
+
     @property
     def supports_partial_writes(self) -> bool:
         # docstring inherited