We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Unable to dvc add more than 1 file at a time in s3 bucket
** Note: DVC version 0.66.1, pip install, Ubuntu 16.04

I am having difficulty configuring DVC to track files on an s3 bucket with the structure

```
ellesdatabucket
       ├── data
       │     ├── img1.png
       │     ├── img2.png
       │     ├── ...
       └── cache
```

Specifically, I want to use DVC to version control *.png files stored in the data folder, and use the cache folder as DVC's cache.

Based on the docs provided [here](https://dvc.org/doc/user-guide/managing-external-data#amazon-s-3), I believe I've replicated exactly the provided steps. But I hit an error when I run `dvc add`:
```
$ git init
$ dvc init
$ dvc remote add myremote s3://ellesdatabucket/data
$ dvc remote add s3cache s3://ellesdatabucket/cache
$ dvc config cache.s3 s3cache
$ dvc add s3://ellesdatabucket/data
```
The output looks initially encouraging,
``` 0%|          |Add                                      0/1 [00:00<?,     ?file/s]
 29%|██▊       |Computing hashes (only done once691/2424 [00:03<00:07,    223md5/s]
```

But then I get this error message:

```100%|██████████|Add                                  1/1 [00:07<00:00,  7.56s/file]
ERROR: s3://ellesdatabucket/data/. does not exist: An error occurred (404) when calling the HeadObject operation: Not Found
```

I'm positive that there are files in the data folder and can view them by `aws s3 ls s3://ellesdatabucket/data`. And, if I try to run `dvc add` with only a single file at a time instead of a whole directory, the command completes successfully. Although a bash script could `dvc add` each file in a loop, I want to make sure there's not a better way. [Issue 2647](https://github.com/iterative/dvc/issues/2647) seems to be discussing a similar problem but I can't figure out how to apply that code to my own example here. 

Thank you for any help!

**More context**:

https://discordapp.com/channels/485586884165107732/485596304961962003/637836708184064010



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dvc/remote/s3.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 from __future__ import unicode_literals
3 
4 import logging
5 import os
6 import threading
7 
8 from funcy import cached_property, wrap_prop
9 
10 from dvc.config import Config
11 from dvc.exceptions import DvcException
12 from dvc.exceptions import ETagMismatchError
13 from dvc.path_info import CloudURLInfo
14 from dvc.progress import Tqdm
15 from dvc.remote.base import RemoteBASE
16 from dvc.scheme import Schemes
17 
18 logger = logging.getLogger(__name__)
19 
20 
21 class RemoteS3(RemoteBASE):
22     scheme = Schemes.S3
23     path_cls = CloudURLInfo
24     REQUIRES = {"boto3": "boto3"}
25     PARAM_CHECKSUM = "etag"
26 
27     def __init__(self, repo, config):
28         super(RemoteS3, self).__init__(repo, config)
29 
30         url = config.get(Config.SECTION_REMOTE_URL, "s3://")
31         self.path_info = self.path_cls(url)
32 
33         self.region = config.get(Config.SECTION_AWS_REGION)
34 
35         self.profile = config.get(Config.SECTION_AWS_PROFILE)
36 
37         self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL)
38 
39         if config.get(Config.SECTION_AWS_LIST_OBJECTS):
40             self.list_objects_api = "list_objects"
41         else:
42             self.list_objects_api = "list_objects_v2"
43 
44         self.use_ssl = config.get(Config.SECTION_AWS_USE_SSL, True)
45 
46         self.extra_args = {}
47 
48         self.sse = config.get(Config.SECTION_AWS_SSE, "")
49         if self.sse:
50             self.extra_args["ServerSideEncryption"] = self.sse
51 
52         self.acl = config.get(Config.SECTION_AWS_ACL, "")
53         if self.acl:
54             self.extra_args["ACL"] = self.acl
55 
56         shared_creds = config.get(Config.SECTION_AWS_CREDENTIALPATH)
57         if shared_creds:
58             os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
59 
60     @wrap_prop(threading.Lock())
61     @cached_property
62     def s3(self):
63         import boto3
64 
65         session = boto3.session.Session(
66             profile_name=self.profile, region_name=self.region
67         )
68 
69         return session.client(
70             "s3", endpoint_url=self.endpoint_url, use_ssl=self.use_ssl
71         )
72 
73     @classmethod
74     def get_etag(cls, s3, bucket, path):
75         obj = cls.get_head_object(s3, bucket, path)
76 
77         return obj["ETag"].strip('"')
78 
79     def get_file_checksum(self, path_info):
80         return self.get_etag(self.s3, path_info.bucket, path_info.path)
81 
82     @staticmethod
83     def get_head_object(s3, bucket, path, *args, **kwargs):
84 
85         try:
86             obj = s3.head_object(Bucket=bucket, Key=path, *args, **kwargs)
87         except Exception as exc:
88             raise DvcException(
89                 "s3://{}/{} does not exist".format(bucket, path), exc
90             )
91         return obj
92 
93     @classmethod
94     def _copy_multipart(
95         cls, s3, from_info, to_info, size, n_parts, extra_args
96     ):
97         mpu = s3.create_multipart_upload(
98             Bucket=to_info.bucket, Key=to_info.path, **extra_args
99         )
100         mpu_id = mpu["UploadId"]
101 
102         parts = []
103         byte_position = 0
104         for i in range(1, n_parts + 1):
105             obj = cls.get_head_object(
106                 s3, from_info.bucket, from_info.path, PartNumber=i
107             )
108             part_size = obj["ContentLength"]
109             lastbyte = byte_position + part_size - 1
110             if lastbyte > size:
111                 lastbyte = size - 1
112 
113             srange = "bytes={}-{}".format(byte_position, lastbyte)
114 
115             part = s3.upload_part_copy(
116                 Bucket=to_info.bucket,
117                 Key=to_info.path,
118                 PartNumber=i,
119                 UploadId=mpu_id,
120                 CopySourceRange=srange,
121                 CopySource={"Bucket": from_info.bucket, "Key": from_info.path},
122             )
123             parts.append(
124                 {"PartNumber": i, "ETag": part["CopyPartResult"]["ETag"]}
125             )
126             byte_position += part_size
127 
128         assert n_parts == len(parts)
129 
130         s3.complete_multipart_upload(
131             Bucket=to_info.bucket,
132             Key=to_info.path,
133             UploadId=mpu_id,
134             MultipartUpload={"Parts": parts},
135         )
136 
137     @classmethod
138     def _copy(cls, s3, from_info, to_info, extra_args):
139         # NOTE: object's etag depends on the way it was uploaded to s3 or the
140         # way it was copied within the s3. More specifically, it depends on
141         # the chunk size that was used to transfer it, which would affect
142         # whether an object would be uploaded as a single part or as a
143         # multipart.
144         #
145         # If an object's etag looks like '8978c98bb5a48c2fb5f2c4c905768afa',
146         # then it was transferred as a single part, which means that the chunk
147         # size used to transfer it was greater or equal to the ContentLength
148         # of that object. So to preserve that tag over the next transfer, we
149         # could use any value >= ContentLength.
150         #
151         # If an object's etag looks like '50d67013a5e1a4070bef1fc8eea4d5f9-13',
152         # then it was transferred as a multipart, which means that the chunk
153         # size used to transfer it was less than ContentLength of that object.
154         # Unfortunately, in general, it doesn't mean that the chunk size was
155         # the same throughout the transfer, so it means that in order to
156         # preserve etag, we need to transfer each part separately, so the
157         # object is transfered in the same chunks as it was originally.
158 
159         obj = cls.get_head_object(s3, from_info.bucket, from_info.path)
160         etag = obj["ETag"].strip('"')
161         size = obj["ContentLength"]
162 
163         _, _, parts_suffix = etag.partition("-")
164         if parts_suffix:
165             n_parts = int(parts_suffix)
166             cls._copy_multipart(
167                 s3, from_info, to_info, size, n_parts, extra_args=extra_args
168             )
169         else:
170             source = {"Bucket": from_info.bucket, "Key": from_info.path}
171             s3.copy(source, to_info.bucket, to_info.path, ExtraArgs=extra_args)
172 
173         cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path)
174         if etag != cached_etag:
175             raise ETagMismatchError(etag, cached_etag)
176 
177     def copy(self, from_info, to_info):
178         self._copy(self.s3, from_info, to_info, self.extra_args)
179 
180     def remove(self, path_info):
181         if path_info.scheme != "s3":
182             raise NotImplementedError
183 
184         logger.debug("Removing {}".format(path_info))
185         self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)
186 
187     def _list_objects(self, path_info, max_items=None):
188         """ Read config for list object api, paginate through list objects."""
189         kwargs = {
190             "Bucket": path_info.bucket,
191             "Prefix": path_info.path,
192             "PaginationConfig": {"MaxItems": max_items},
193         }
194         paginator = self.s3.get_paginator(self.list_objects_api)
195         for page in paginator.paginate(**kwargs):
196             contents = page.get("Contents", None)
197             if not contents:
198                 continue
199             for item in contents:
200                 yield item
201 
202     def _list_paths(self, path_info, max_items=None):
203         return (
204             item["Key"] for item in self._list_objects(path_info, max_items)
205         )
206 
207     def list_cache_paths(self):
208         return self._list_paths(self.path_info)
209 
210     def exists(self, path_info):
211         dir_path = path_info / ""
212         fname = next(self._list_paths(path_info, max_items=1), "")
213         return path_info.path == fname or fname.startswith(dir_path.path)
214 
215     def isdir(self, path_info):
216         # S3 doesn't have a concept for directories.
217         #
218         # Using `head_object` with a path pointing to a directory
219         # will throw a 404 error.
220         #
221         # A reliable way to know if a given path is a directory is by
222         # checking if there are more files sharing the same prefix
223         # with a `list_objects` call.
224         #
225         # We need to make sure that the path ends with a forward slash,
226         # since we can end with false-positives like the following example:
227         #
228         #       bucket
229         #       └── data
230         #          ├── alice
231         #          └── alpha
232         #
233         # Using `data/al` as prefix will return `[data/alice, data/alpha]`,
234         # While `data/al/` will return nothing.
235         #
236         dir_path = path_info / ""
237         return bool(list(self._list_paths(dir_path, max_items=1)))
238 
239     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
240         total = os.path.getsize(from_file)
241         with Tqdm(
242             disable=no_progress_bar, total=total, bytes=True, desc=name
243         ) as pbar:
244             self.s3.upload_file(
245                 from_file,
246                 to_info.bucket,
247                 to_info.path,
248                 Callback=pbar.update,
249                 ExtraArgs=self.extra_args,
250             )
251 
252     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
253         if no_progress_bar:
254             total = None
255         else:
256             total = self.s3.head_object(
257                 Bucket=from_info.bucket, Key=from_info.path
258             )["ContentLength"]
259         with Tqdm(
260             disable=no_progress_bar, total=total, bytes=True, desc=name
261         ) as pbar:
262             self.s3.download_file(
263                 from_info.bucket, from_info.path, to_file, Callback=pbar.update
264             )
265 
266     def _generate_download_url(self, path_info, expires=3600):
267         params = {"Bucket": path_info.bucket, "Key": path_info.path}
268         return self.s3.generate_presigned_url(
269             ClientMethod="get_object", Params=params, ExpiresIn=int(expires)
270         )
271 
272     def walk_files(self, path_info, max_items=None):
273         for fname in self._list_paths(path_info, max_items):
274             yield path_info.replace(path=fname)
275
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -212,6 +212,15 @@
         fname = next(self._list_paths(path_info, max_items=1), "")
         return path_info.path == fname or fname.startswith(dir_path.path)
 
+    def makedirs(self, path_info):
+        # We need to support creating empty directories, which means
+        # creating an object with an empty body and a trailing slash `/`.
+        #
+        # We are not creating directory objects for every parent prefix,
+        # as it is not required.
+        dir_path = path_info / ""
+        self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body="")
+
     def isdir(self, path_info):
         # S3 doesn't have a concept for directories.
         #
@@ -271,4 +280,7 @@
 
     def walk_files(self, path_info, max_items=None):
         for fname in self._list_paths(path_info, max_items):
+            if fname.endswith("/"):
+                continue
+
             yield path_info.replace(path=fname)

{"golden_diff": "diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py\n--- a/dvc/remote/s3.py\n+++ b/dvc/remote/s3.py\n@@ -212,6 +212,15 @@\n         fname = next(self._list_paths(path_info, max_items=1), \"\")\n         return path_info.path == fname or fname.startswith(dir_path.path)\n \n+    def makedirs(self, path_info):\n+        # We need to support creating empty directories, which means\n+        # creating an object with an empty body and a trailing slash `/`.\n+        #\n+        # We are not creating directory objects for every parent prefix,\n+        # as it is not required.\n+        dir_path = path_info / \"\"\n+        self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body=\"\")\n+\n     def isdir(self, path_info):\n         # S3 doesn't have a concept for directories.\n         #\n@@ -271,4 +280,7 @@\n \n     def walk_files(self, path_info, max_items=None):\n         for fname in self._list_paths(path_info, max_items):\n+            if fname.endswith(\"/\"):\n+                continue\n+\n             yield path_info.replace(path=fname)\n", "issue": "Unable to dvc add more than 1 file at a time in s3 bucket\n** Note: DVC version 0.66.1, pip install, Ubuntu 16.04\r\n\r\nI am having difficulty configuring DVC to track files on an s3 bucket with the structure\r\n\r\n```\r\nellesdatabucket\r\n       \u251c\u2500\u2500 data\r\n       \u2502     \u251c\u2500\u2500 img1.png\r\n       \u2502     \u251c\u2500\u2500 img2.png\r\n       \u2502     \u251c\u2500\u2500 ...\r\n       \u2514\u2500\u2500 cache\r\n```\r\n\r\nSpecifically, I want to use DVC to version control *.png files stored in the data folder, and use the cache folder as DVC's cache.\r\n\r\nBased on the docs provided [here](https://dvc.org/doc/user-guide/managing-external-data#amazon-s-3), I believe I've replicated exactly the provided steps. But I hit an error when I run `dvc add`:\r\n```\r\n$ git init\r\n$ dvc init\r\n$ dvc remote add myremote s3://ellesdatabucket/data\r\n$ dvc remote add s3cache s3://ellesdatabucket/cache\r\n$ dvc config cache.s3 s3cache\r\n$ dvc add s3://ellesdatabucket/data\r\n```\r\nThe output looks initially encouraging,\r\n``` 0%|          |Add                                      0/1 [00:00<?,     ?file/s]\r\n 29%|\u2588\u2588\u258a       |Computing hashes (only done once691/2424 [00:03<00:07,    223md5/s]\r\n```\r\n\r\nBut then I get this error message:\r\n\r\n```100%|\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588|Add                                  1/1 [00:07<00:00,  7.56s/file]\r\nERROR: s3://ellesdatabucket/data/. does not exist: An error occurred (404) when calling the HeadObject operation: Not Found\r\n```\r\n\r\nI'm positive that there are files in the data folder and can view them by `aws s3 ls s3://ellesdatabucket/data`. And, if I try to run `dvc add` with only a single file at a time instead of a whole directory, the command completes successfully. Although a bash script could `dvc add` each file in a loop, I want to make sure there's not a better way. [Issue 2647](https://github.com/iterative/dvc/issues/2647) seems to be discussing a similar problem but I can't figure out how to apply that code to my own example here. \r\n\r\nThank you for any help!\r\n\r\n**More context**:\r\n\r\nhttps://discordapp.com/channels/485586884165107732/485596304961962003/637836708184064010\r\n\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\nfrom __future__ import unicode_literals\n\nimport logging\nimport os\nimport threading\n\nfrom funcy import cached_property, wrap_prop\n\nfrom dvc.config import Config\nfrom dvc.exceptions import DvcException\nfrom dvc.exceptions import ETagMismatchError\nfrom dvc.path_info import CloudURLInfo\nfrom dvc.progress import Tqdm\nfrom dvc.remote.base import RemoteBASE\nfrom dvc.scheme import Schemes\n\nlogger = logging.getLogger(__name__)\n\n\nclass RemoteS3(RemoteBASE):\n    scheme = Schemes.S3\n    path_cls = CloudURLInfo\n    REQUIRES = {\"boto3\": \"boto3\"}\n    PARAM_CHECKSUM = \"etag\"\n\n    def __init__(self, repo, config):\n        super(RemoteS3, self).__init__(repo, config)\n\n        url = config.get(Config.SECTION_REMOTE_URL, \"s3://\")\n        self.path_info = self.path_cls(url)\n\n        self.region = config.get(Config.SECTION_AWS_REGION)\n\n        self.profile = config.get(Config.SECTION_AWS_PROFILE)\n\n        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL)\n\n        if config.get(Config.SECTION_AWS_LIST_OBJECTS):\n            self.list_objects_api = \"list_objects\"\n        else:\n            self.list_objects_api = \"list_objects_v2\"\n\n        self.use_ssl = config.get(Config.SECTION_AWS_USE_SSL, True)\n\n        self.extra_args = {}\n\n        self.sse = config.get(Config.SECTION_AWS_SSE, \"\")\n        if self.sse:\n            self.extra_args[\"ServerSideEncryption\"] = self.sse\n\n        self.acl = config.get(Config.SECTION_AWS_ACL, \"\")\n        if self.acl:\n            self.extra_args[\"ACL\"] = self.acl\n\n        shared_creds = config.get(Config.SECTION_AWS_CREDENTIALPATH)\n        if shared_creds:\n            os.environ.setdefault(\"AWS_SHARED_CREDENTIALS_FILE\", shared_creds)\n\n    @wrap_prop(threading.Lock())\n    @cached_property\n    def s3(self):\n        import boto3\n\n        session = boto3.session.Session(\n            profile_name=self.profile, region_name=self.region\n        )\n\n        return session.client(\n            \"s3\", endpoint_url=self.endpoint_url, use_ssl=self.use_ssl\n        )\n\n    @classmethod\n    def get_etag(cls, s3, bucket, path):\n        obj = cls.get_head_object(s3, bucket, path)\n\n        return obj[\"ETag\"].strip('\"')\n\n    def get_file_checksum(self, path_info):\n        return self.get_etag(self.s3, path_info.bucket, path_info.path)\n\n    @staticmethod\n    def get_head_object(s3, bucket, path, *args, **kwargs):\n\n        try:\n            obj = s3.head_object(Bucket=bucket, Key=path, *args, **kwargs)\n        except Exception as exc:\n            raise DvcException(\n                \"s3://{}/{} does not exist\".format(bucket, path), exc\n            )\n        return obj\n\n    @classmethod\n    def _copy_multipart(\n        cls, s3, from_info, to_info, size, n_parts, extra_args\n    ):\n        mpu = s3.create_multipart_upload(\n            Bucket=to_info.bucket, Key=to_info.path, **extra_args\n        )\n        mpu_id = mpu[\"UploadId\"]\n\n        parts = []\n        byte_position = 0\n        for i in range(1, n_parts + 1):\n            obj = cls.get_head_object(\n                s3, from_info.bucket, from_info.path, PartNumber=i\n            )\n            part_size = obj[\"ContentLength\"]\n            lastbyte = byte_position + part_size - 1\n            if lastbyte > size:\n                lastbyte = size - 1\n\n            srange = \"bytes={}-{}\".format(byte_position, lastbyte)\n\n            part = s3.upload_part_copy(\n                Bucket=to_info.bucket,\n                Key=to_info.path,\n                PartNumber=i,\n                UploadId=mpu_id,\n                CopySourceRange=srange,\n                CopySource={\"Bucket\": from_info.bucket, \"Key\": from_info.path},\n            )\n            parts.append(\n                {\"PartNumber\": i, \"ETag\": part[\"CopyPartResult\"][\"ETag\"]}\n            )\n            byte_position += part_size\n\n        assert n_parts == len(parts)\n\n        s3.complete_multipart_upload(\n            Bucket=to_info.bucket,\n            Key=to_info.path,\n            UploadId=mpu_id,\n            MultipartUpload={\"Parts\": parts},\n        )\n\n    @classmethod\n    def _copy(cls, s3, from_info, to_info, extra_args):\n        # NOTE: object's etag depends on the way it was uploaded to s3 or the\n        # way it was copied within the s3. More specifically, it depends on\n        # the chunk size that was used to transfer it, which would affect\n        # whether an object would be uploaded as a single part or as a\n        # multipart.\n        #\n        # If an object's etag looks like '8978c98bb5a48c2fb5f2c4c905768afa',\n        # then it was transferred as a single part, which means that the chunk\n        # size used to transfer it was greater or equal to the ContentLength\n        # of that object. So to preserve that tag over the next transfer, we\n        # could use any value >= ContentLength.\n        #\n        # If an object's etag looks like '50d67013a5e1a4070bef1fc8eea4d5f9-13',\n        # then it was transferred as a multipart, which means that the chunk\n        # size used to transfer it was less than ContentLength of that object.\n        # Unfortunately, in general, it doesn't mean that the chunk size was\n        # the same throughout the transfer, so it means that in order to\n        # preserve etag, we need to transfer each part separately, so the\n        # object is transfered in the same chunks as it was originally.\n\n        obj = cls.get_head_object(s3, from_info.bucket, from_info.path)\n        etag = obj[\"ETag\"].strip('\"')\n        size = obj[\"ContentLength\"]\n\n        _, _, parts_suffix = etag.partition(\"-\")\n        if parts_suffix:\n            n_parts = int(parts_suffix)\n            cls._copy_multipart(\n                s3, from_info, to_info, size, n_parts, extra_args=extra_args\n            )\n        else:\n            source = {\"Bucket\": from_info.bucket, \"Key\": from_info.path}\n            s3.copy(source, to_info.bucket, to_info.path, ExtraArgs=extra_args)\n\n        cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path)\n        if etag != cached_etag:\n            raise ETagMismatchError(etag, cached_etag)\n\n    def copy(self, from_info, to_info):\n        self._copy(self.s3, from_info, to_info, self.extra_args)\n\n    def remove(self, path_info):\n        if path_info.scheme != \"s3\":\n            raise NotImplementedError\n\n        logger.debug(\"Removing {}\".format(path_info))\n        self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)\n\n    def _list_objects(self, path_info, max_items=None):\n        \"\"\" Read config for list object api, paginate through list objects.\"\"\"\n        kwargs = {\n            \"Bucket\": path_info.bucket,\n            \"Prefix\": path_info.path,\n            \"PaginationConfig\": {\"MaxItems\": max_items},\n        }\n        paginator = self.s3.get_paginator(self.list_objects_api)\n        for page in paginator.paginate(**kwargs):\n            contents = page.get(\"Contents\", None)\n            if not contents:\n                continue\n            for item in contents:\n                yield item\n\n    def _list_paths(self, path_info, max_items=None):\n        return (\n            item[\"Key\"] for item in self._list_objects(path_info, max_items)\n        )\n\n    def list_cache_paths(self):\n        return self._list_paths(self.path_info)\n\n    def exists(self, path_info):\n        dir_path = path_info / \"\"\n        fname = next(self._list_paths(path_info, max_items=1), \"\")\n        return path_info.path == fname or fname.startswith(dir_path.path)\n\n    def isdir(self, path_info):\n        # S3 doesn't have a concept for directories.\n        #\n        # Using `head_object` with a path pointing to a directory\n        # will throw a 404 error.\n        #\n        # A reliable way to know if a given path is a directory is by\n        # checking if there are more files sharing the same prefix\n        # with a `list_objects` call.\n        #\n        # We need to make sure that the path ends with a forward slash,\n        # since we can end with false-positives like the following example:\n        #\n        #       bucket\n        #       \u2514\u2500\u2500 data\n        #          \u251c\u2500\u2500 alice\n        #          \u2514\u2500\u2500 alpha\n        #\n        # Using `data/al` as prefix will return `[data/alice, data/alpha]`,\n        # While `data/al/` will return nothing.\n        #\n        dir_path = path_info / \"\"\n        return bool(list(self._list_paths(dir_path, max_items=1)))\n\n    def _upload(self, from_file, to_info, name=None, no_progress_bar=False):\n        total = os.path.getsize(from_file)\n        with Tqdm(\n            disable=no_progress_bar, total=total, bytes=True, desc=name\n        ) as pbar:\n            self.s3.upload_file(\n                from_file,\n                to_info.bucket,\n                to_info.path,\n                Callback=pbar.update,\n                ExtraArgs=self.extra_args,\n            )\n\n    def _download(self, from_info, to_file, name=None, no_progress_bar=False):\n        if no_progress_bar:\n            total = None\n        else:\n            total = self.s3.head_object(\n                Bucket=from_info.bucket, Key=from_info.path\n            )[\"ContentLength\"]\n        with Tqdm(\n            disable=no_progress_bar, total=total, bytes=True, desc=name\n        ) as pbar:\n            self.s3.download_file(\n                from_info.bucket, from_info.path, to_file, Callback=pbar.update\n            )\n\n    def _generate_download_url(self, path_info, expires=3600):\n        params = {\"Bucket\": path_info.bucket, \"Key\": path_info.path}\n        return self.s3.generate_presigned_url(\n            ClientMethod=\"get_object\", Params=params, ExpiresIn=int(expires)\n        )\n\n    def walk_files(self, path_info, max_items=None):\n        for fname in self._list_paths(path_info, max_items):\n            yield path_info.replace(path=fname)\n", "path": "dvc/remote/s3.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\nfrom __future__ import unicode_literals\n\nimport logging\nimport os\nimport threading\n\nfrom funcy import cached_property, wrap_prop\n\nfrom dvc.config import Config\nfrom dvc.exceptions import DvcException\nfrom dvc.exceptions import ETagMismatchError\nfrom dvc.path_info import CloudURLInfo\nfrom dvc.progress import Tqdm\nfrom dvc.remote.base import RemoteBASE\nfrom dvc.scheme import Schemes\n\nlogger = logging.getLogger(__name__)\n\n\nclass RemoteS3(RemoteBASE):\n    scheme = Schemes.S3\n    path_cls = CloudURLInfo\n    REQUIRES = {\"boto3\": \"boto3\"}\n    PARAM_CHECKSUM = \"etag\"\n\n    def __init__(self, repo, config):\n        super(RemoteS3, self).__init__(repo, config)\n\n        url = config.get(Config.SECTION_REMOTE_URL, \"s3://\")\n        self.path_info = self.path_cls(url)\n\n        self.region = config.get(Config.SECTION_AWS_REGION)\n\n        self.profile = config.get(Config.SECTION_AWS_PROFILE)\n\n        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL)\n\n        if config.get(Config.SECTION_AWS_LIST_OBJECTS):\n            self.list_objects_api = \"list_objects\"\n        else:\n            self.list_objects_api = \"list_objects_v2\"\n\n        self.use_ssl = config.get(Config.SECTION_AWS_USE_SSL, True)\n\n        self.extra_args = {}\n\n        self.sse = config.get(Config.SECTION_AWS_SSE, \"\")\n        if self.sse:\n            self.extra_args[\"ServerSideEncryption\"] = self.sse\n\n        self.acl = config.get(Config.SECTION_AWS_ACL, \"\")\n        if self.acl:\n            self.extra_args[\"ACL\"] = self.acl\n\n        shared_creds = config.get(Config.SECTION_AWS_CREDENTIALPATH)\n        if shared_creds:\n            os.environ.setdefault(\"AWS_SHARED_CREDENTIALS_FILE\", shared_creds)\n\n    @wrap_prop(threading.Lock())\n    @cached_property\n    def s3(self):\n        import boto3\n\n        session = boto3.session.Session(\n            profile_name=self.profile, region_name=self.region\n        )\n\n        return session.client(\n            \"s3\", endpoint_url=self.endpoint_url, use_ssl=self.use_ssl\n        )\n\n    @classmethod\n    def get_etag(cls, s3, bucket, path):\n        obj = cls.get_head_object(s3, bucket, path)\n\n        return obj[\"ETag\"].strip('\"')\n\n    def get_file_checksum(self, path_info):\n        return self.get_etag(self.s3, path_info.bucket, path_info.path)\n\n    @staticmethod\n    def get_head_object(s3, bucket, path, *args, **kwargs):\n\n        try:\n            obj = s3.head_object(Bucket=bucket, Key=path, *args, **kwargs)\n        except Exception as exc:\n            raise DvcException(\n                \"s3://{}/{} does not exist\".format(bucket, path), exc\n            )\n        return obj\n\n    @classmethod\n    def _copy_multipart(\n        cls, s3, from_info, to_info, size, n_parts, extra_args\n    ):\n        mpu = s3.create_multipart_upload(\n            Bucket=to_info.bucket, Key=to_info.path, **extra_args\n        )\n        mpu_id = mpu[\"UploadId\"]\n\n        parts = []\n        byte_position = 0\n        for i in range(1, n_parts + 1):\n            obj = cls.get_head_object(\n                s3, from_info.bucket, from_info.path, PartNumber=i\n            )\n            part_size = obj[\"ContentLength\"]\n            lastbyte = byte_position + part_size - 1\n            if lastbyte > size:\n                lastbyte = size - 1\n\n            srange = \"bytes={}-{}\".format(byte_position, lastbyte)\n\n            part = s3.upload_part_copy(\n                Bucket=to_info.bucket,\n                Key=to_info.path,\n                PartNumber=i,\n                UploadId=mpu_id,\n                CopySourceRange=srange,\n                CopySource={\"Bucket\": from_info.bucket, \"Key\": from_info.path},\n            )\n            parts.append(\n                {\"PartNumber\": i, \"ETag\": part[\"CopyPartResult\"][\"ETag\"]}\n            )\n            byte_position += part_size\n\n        assert n_parts == len(parts)\n\n        s3.complete_multipart_upload(\n            Bucket=to_info.bucket,\n            Key=to_info.path,\n            UploadId=mpu_id,\n            MultipartUpload={\"Parts\": parts},\n        )\n\n    @classmethod\n    def _copy(cls, s3, from_info, to_info, extra_args):\n        # NOTE: object's etag depends on the way it was uploaded to s3 or the\n        # way it was copied within the s3. More specifically, it depends on\n        # the chunk size that was used to transfer it, which would affect\n        # whether an object would be uploaded as a single part or as a\n        # multipart.\n        #\n        # If an object's etag looks like '8978c98bb5a48c2fb5f2c4c905768afa',\n        # then it was transferred as a single part, which means that the chunk\n        # size used to transfer it was greater or equal to the ContentLength\n        # of that object. So to preserve that tag over the next transfer, we\n        # could use any value >= ContentLength.\n        #\n        # If an object's etag looks like '50d67013a5e1a4070bef1fc8eea4d5f9-13',\n        # then it was transferred as a multipart, which means that the chunk\n        # size used to transfer it was less than ContentLength of that object.\n        # Unfortunately, in general, it doesn't mean that the chunk size was\n        # the same throughout the transfer, so it means that in order to\n        # preserve etag, we need to transfer each part separately, so the\n        # object is transfered in the same chunks as it was originally.\n\n        obj = cls.get_head_object(s3, from_info.bucket, from_info.path)\n        etag = obj[\"ETag\"].strip('\"')\n        size = obj[\"ContentLength\"]\n\n        _, _, parts_suffix = etag.partition(\"-\")\n        if parts_suffix:\n            n_parts = int(parts_suffix)\n            cls._copy_multipart(\n                s3, from_info, to_info, size, n_parts, extra_args=extra_args\n            )\n        else:\n            source = {\"Bucket\": from_info.bucket, \"Key\": from_info.path}\n            s3.copy(source, to_info.bucket, to_info.path, ExtraArgs=extra_args)\n\n        cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path)\n        if etag != cached_etag:\n            raise ETagMismatchError(etag, cached_etag)\n\n    def copy(self, from_info, to_info):\n        self._copy(self.s3, from_info, to_info, self.extra_args)\n\n    def remove(self, path_info):\n        if path_info.scheme != \"s3\":\n            raise NotImplementedError\n\n        logger.debug(\"Removing {}\".format(path_info))\n        self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)\n\n    def _list_objects(self, path_info, max_items=None):\n        \"\"\" Read config for list object api, paginate through list objects.\"\"\"\n        kwargs = {\n            \"Bucket\": path_info.bucket,\n            \"Prefix\": path_info.path,\n            \"PaginationConfig\": {\"MaxItems\": max_items},\n        }\n        paginator = self.s3.get_paginator(self.list_objects_api)\n        for page in paginator.paginate(**kwargs):\n            contents = page.get(\"Contents\", None)\n            if not contents:\n                continue\n            for item in contents:\n                yield item\n\n    def _list_paths(self, path_info, max_items=None):\n        return (\n            item[\"Key\"] for item in self._list_objects(path_info, max_items)\n        )\n\n    def list_cache_paths(self):\n        return self._list_paths(self.path_info)\n\n    def exists(self, path_info):\n        dir_path = path_info / \"\"\n        fname = next(self._list_paths(path_info, max_items=1), \"\")\n        return path_info.path == fname or fname.startswith(dir_path.path)\n\n    def makedirs(self, path_info):\n        # We need to support creating empty directories, which means\n        # creating an object with an empty body and a trailing slash `/`.\n        #\n        # We are not creating directory objects for every parent prefix,\n        # as it is not required.\n        dir_path = path_info / \"\"\n        self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body=\"\")\n\n    def isdir(self, path_info):\n        # S3 doesn't have a concept for directories.\n        #\n        # Using `head_object` with a path pointing to a directory\n        # will throw a 404 error.\n        #\n        # A reliable way to know if a given path is a directory is by\n        # checking if there are more files sharing the same prefix\n        # with a `list_objects` call.\n        #\n        # We need to make sure that the path ends with a forward slash,\n        # since we can end with false-positives like the following example:\n        #\n        #       bucket\n        #       \u2514\u2500\u2500 data\n        #          \u251c\u2500\u2500 alice\n        #          \u2514\u2500\u2500 alpha\n        #\n        # Using `data/al` as prefix will return `[data/alice, data/alpha]`,\n        # While `data/al/` will return nothing.\n        #\n        dir_path = path_info / \"\"\n        return bool(list(self._list_paths(dir_path, max_items=1)))\n\n    def _upload(self, from_file, to_info, name=None, no_progress_bar=False):\n        total = os.path.getsize(from_file)\n        with Tqdm(\n            disable=no_progress_bar, total=total, bytes=True, desc=name\n        ) as pbar:\n            self.s3.upload_file(\n                from_file,\n                to_info.bucket,\n                to_info.path,\n                Callback=pbar.update,\n                ExtraArgs=self.extra_args,\n            )\n\n    def _download(self, from_info, to_file, name=None, no_progress_bar=False):\n        if no_progress_bar:\n            total = None\n        else:\n            total = self.s3.head_object(\n                Bucket=from_info.bucket, Key=from_info.path\n            )[\"ContentLength\"]\n        with Tqdm(\n            disable=no_progress_bar, total=total, bytes=True, desc=name\n        ) as pbar:\n            self.s3.download_file(\n                from_info.bucket, from_info.path, to_file, Callback=pbar.update\n            )\n\n    def _generate_download_url(self, path_info, expires=3600):\n        params = {\"Bucket\": path_info.bucket, \"Key\": path_info.path}\n        return self.s3.generate_presigned_url(\n            ClientMethod=\"get_object\", Params=params, ExpiresIn=int(expires)\n        )\n\n    def walk_files(self, path_info, max_items=None):\n        for fname in self._list_paths(path_info, max_items):\n            if fname.endswith(\"/\"):\n                continue\n\n            yield path_info.replace(path=fname)\n", "path": "dvc/remote/s3.py"}]}