Merge pull request #41 from dask/bulk_delete

PR: Add bulk delete

s3fs/core.py

@@ -454,6 +454,32 @@ def copy(self, path1, path2):
             raise IOError('Copy failed', (path1, path2))
         self.invalidate_cache(buc2)
 
+    def bulk_delete(self, pathlist):
+        """
+        Remove multiple keys with one call
+
+        Parameters
+        ----------
+        pathlist : listof strings
+            The keys to remove, must all be in the same bucket.
+        """
+        buckets = {split_path(path)[0] for path in pathlist}
+        if len(buckets) > 1:
+            raise ValueError("Bulk delete files should refer to only one bucket")
+        bucket = buckets.pop()
+        if len(pathlist) > 1000:
+            for i in range((len(pathlist) // 1000) + 1):
+                print(i)
+                self.bulk_delete(pathlist[i*1000:(i+1)*1000])
+            return
+        delete_keys = {'Objects': [{'Key' : split_path(path)[1]} for path
+                                   in pathlist]}
+        try:
+            self.s3.delete_objects(Bucket=bucket, Delete=delete_keys)
+            self.invalidate_cache(bucket)
+        except ClientError:
+            raise IOError('Bulk delete failed')
+
     def rm(self, path, recursive=False):
         """
         Remove keys and/or bucket.
@@ -469,9 +495,9 @@ def rm(self, path, recursive=False):
         if not self.exists(path):
             raise FileNotFoundError(path)
         if recursive:
-            for f in self.walk(path):
-                self.rm(f, recursive=False)
-            return
+            self.bulk_delete(self.walk(path))
+            if not self.exists(path):
+                return
         bucket, key = split_path(path)
         if key:
             try:

s3fs/tests/test_s3fs.py

@@ -159,6 +159,22 @@ def test_rm(s3):
     s3.rm(test_bucket_name+'/nested', recursive=True)
     assert not s3.exists(test_bucket_name+'/nested/nested2/file1')
 
+    #whole bucket
+    s3.rm(test_bucket_name, recursive=True)
+    with pytest.raises((IOError, OSError)):
+        s3.exists(test_bucket_name+'/2014-01-01.csv')
+    assert not s3.exists(test_bucket_name)
+
+
+def test_bulk_delete(s3):
+    with pytest.raises((OSError, IOError)):
+        s3.bulk_delete(['nonexistent/file'])
+    with pytest.raises(ValueError):
+        s3.bulk_delete(['bucket1/file', 'bucket2/file'])
+    filelist = s3.walk(test_bucket_name+'/nested')
+    s3.bulk_delete(filelist)
+    assert not s3.exists(test_bucket_name+'/nested/nested2/file1')
+
 
 def test_anonymous_access():
     with ignoring(NoCredentialsError):
@@ -212,6 +228,7 @@ def test_s3_big_ls(s3):
         s3.touch(test_bucket_name+'/thousand/%i.part'%x)
     assert len(s3.walk(test_bucket_name)) > 1200
     s3.rm(test_bucket_name+'/thousand/', recursive=True)
+    assert len(s3.walk(test_bucket_name+'/thousand/')) == 0
 
 
 def test_s3_ls_detail(s3):