Merge pull request #374 from martindurant/xarray_test

Xarray test via dask threads

Author: martindurant

s3fs/core.py

@@ -204,7 +204,9 @@ async def _call_s3(self, method, *akwarglist, **kwargs):
                                                        **kwargs)
         for i in range(self.retries):
             try:
-                return await method(**additional_kwargs)
+                out = await method(**additional_kwargs)
+                locals().pop("err", None)  # break cycle following retry
+                return out
             except S3_RETRYABLE_ERRORS as e:
                 logger.debug("Retryable error: %s" % e)
                 err = e
@@ -484,7 +486,7 @@ async def _find(self, path, maxdepth=None, withdirs=None, detail=False):
     async def _mkdir(self, path, acl="", create_parents=True, **kwargs):
         path = self._strip_protocol(path).rstrip('/')
         bucket, key, _ = self.split_path(path)
-        if not key or (create_parents and not self.exists(bucket)):
+        if not key or (create_parents and not await self._exists(bucket)):
             if acl and acl not in buck_acls:
                 raise ValueError('ACL not in %s', buck_acls)
             try:
@@ -595,6 +597,12 @@ async def _exists(self, path):
         elif self.dircache.get(bucket, False):
             return True
         else:
+            try:
+                if self._ls_from_cache(bucket):
+                    return True
+            except FileNotFoundError:
+                # might still be a bucket we can access but don't own
+                pass
             try:
                 await self.s3.list_objects_v2(MaxKeys=1, Bucket=bucket, **self.req_kw)
                 return True

s3fs/tests/test_s3fs.py

@@ -1641,6 +1641,20 @@ def test_repeat_exists(s3):
     assert s3.exists(fn)
 
 
+def test_with_xzarr(s3):
+    da = pytest.importorskip("dask.array")
+    xr = pytest.importorskip("xarray")
+    name = "sample"
+
+    nana = xr.DataArray(da.random.random((1024, 1024, 10, 9, 1)))
+
+    s3_path = f"{test_bucket_name}/{name}"
+    s3store = s3.get_mapper(s3_path)
+
+    s3.ls("")
+    nana.to_dataset().to_zarr(store=s3store, mode="w", consolidated=True, compute=True)
+
+
 @pytest.mark.skipif(sys.version_info < (3, 7), reason="no asyncio.run in py36")
 def test_async_close():
     async def _():