obstore delete_dir (#3310)

* obstore delete_dir

* add cl

* add a delete test

* use obstore list and collect_async

* delete test

---------

Co-authored-by: Davis Bennett <[email protected]>

Author: slowjazz

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:
@@ -195,6 +196,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_listing(self) -> bool:
         # docstring inherited