We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
s3: Support AWS credential settings from environment variables
dvc should also load credential information from system environment variables as awscli does.

Reference: https://docs.aws.amazon.com/cli/latest/userguide/cli-environment.html

For example: we should be able to run `dvc pull` despite absence of AWS profiles in _~/.aws_ if we had set the following environment variables:

```
$ export AWS_ACCESS_KEY_ID=somekey
$ export AWS_SECRET_ACCESS_KEY=somekey
$ export AWS_DEFAULT_REGION=someregion
```

The following command works:
``` 
aws s3 ls s3://bucketname
```

But this one does not:
```
dvc pull
```

A usecase for this is doing `dvc pull` in docker image building process, which would be much easier if if we can override configuration via environment variables. 

--- 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 import os
2 import threading
3 import posixpath
4 
5 try:
6     import boto3
7 except ImportError:
8     boto3 = None
9 
10 try:
11     from urlparse import urlparse
12 except ImportError:
13     from urllib.parse import urlparse
14 
15 from dvc.logger import Logger
16 from dvc.progress import progress
17 from dvc.config import Config
18 from dvc.remote.base import RemoteBase
19 from dvc.remote.local import RemoteLOCAL
20 from dvc.exceptions import DvcException
21 
22 
23 class Callback(object):
24     def __init__(self, name, total):
25         self.name = name
26         self.total = total
27         self.current = 0
28         self.lock = threading.Lock()
29 
30     def __call__(self, byts):
31         with self.lock:
32             self.current += byts
33             progress.update_target(self.name, self.current, self.total)
34 
35 
36 class RemoteS3(RemoteBase):
37     scheme = 's3'
38     REGEX = r'^s3://(?P<path>.*)$'
39     REQUIRES = {'boto3': boto3}
40     PARAM_ETAG = 'etag'
41 
42     def __init__(self, project, config):
43         import configobj
44 
45         self.project = project
46         storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH,
47                                            '').lstrip('/')
48         self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
49         self.region = config.get(Config.SECTION_AWS_REGION, None)
50         self.profile = config.get(Config.SECTION_AWS_PROFILE, 'default')
51         self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)
52 
53         credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)
54         if credentialpath:
55             creds_conf = configobj.ConfigObj(credentialpath)
56             creds = creds_conf.get(self.profile, {})
57         else:
58             creds = {}
59 
60         self.creds = creds
61         self.region = creds.get('region', self.region)
62         self.aws_access_key_id = creds.get('aws_access_key_id', None)
63         self.aws_secret_access_key = creds.get('aws_secret_access_key', None)
64 
65     @property
66     def bucket(self):
67         return urlparse(self.url).netloc
68 
69     @property
70     def prefix(self):
71         return urlparse(self.url).path.lstrip('/')
72 
73     @property
74     def s3(self):
75         if not self.creds:
76             session = boto3.session.Session(profile_name=self.profile)
77         else:
78             session = boto3.session.Session(
79                              aws_access_key_id=self.aws_access_key_id,
80                              aws_secret_access_key=self.aws_secret_access_key,
81                              region_name=self.region)
82         return session.client('s3', endpoint_url=self.endpoint_url)
83 
84     def get_etag(self, bucket, key):
85         try:
86             obj = self.s3.head_object(Bucket=bucket, Key=key)
87         except Exception:
88             raise DvcException('s3://{}/{} does not exist'.format(bucket, key))
89 
90         return obj['ETag'].strip('"')
91 
92     def save_info(self, path_info):
93         if path_info['scheme'] != 's3':
94             raise NotImplementedError
95 
96         return {self.PARAM_ETAG: self.get_etag(path_info['bucket'],
97                                                path_info['key'])}
98 
99     def changed(self, path_info, checksum_info):
100         if not self.exists([path_info])[0]:
101             return True
102 
103         etag = checksum_info.get(self.PARAM_ETAG, None)
104         if etag is None:
105             return True
106 
107         if self.changed_cache(etag):
108             return True
109 
110         return checksum_info != self.save_info(path_info)
111 
112     def _copy(self, from_info, to_info, s3=None):
113         s3 = s3 if s3 else self.s3
114 
115         source = {'Bucket': from_info['bucket'],
116                   'Key': from_info['key']}
117         self.s3.copy(source, to_info['bucket'], to_info['key'])
118 
119     def save(self, path_info):
120         if path_info['scheme'] != 's3':
121             raise NotImplementedError
122 
123         etag = self.get_etag(path_info['bucket'], path_info['key'])
124         key = posixpath.join(self.prefix, etag[0:2], etag[2:])
125         to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
126 
127         self._copy(path_info, to_info)
128 
129         return {self.PARAM_ETAG: etag}
130 
131     @staticmethod
132     def to_string(path_info):
133         return "s3://{}/{}".format(path_info['bucket'], path_info['key'])
134 
135     def changed_cache(self, etag):
136         key = posixpath.join(self.prefix, etag[0:2], etag[2:])
137         cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
138 
139         if {self.PARAM_ETAG: etag} != self.save_info(cache):
140             if self.exists([cache])[0]:
141                 msg = 'Corrupted cache file {}'
142                 Logger.warn(msg.format(self.to_string(cache)))
143                 self.remove(cache)
144             return True
145 
146         return False
147 
148     def checkout(self, path_info, checksum_info):
149         if path_info['scheme'] != 's3':
150             raise NotImplementedError
151 
152         etag = checksum_info.get(self.PARAM_ETAG, None)
153         if not etag:
154             return
155 
156         if not self.changed(path_info, checksum_info):
157             msg = "Data '{}' didn't change."
158             Logger.info(msg.format(self.to_string(path_info)))
159             return
160 
161         if self.changed_cache(etag):
162             msg = "Cache '{}' not found. File '{}' won't be created."
163             Logger.warn(msg.format(etag, self.to_string(path_info)))
164             return
165 
166         if self.exists([path_info])[0]:
167             msg = "Data '{}' exists. Removing before checkout."
168             Logger.warn(msg.format(self.to_string(path_info)))
169             self.remove(path_info)
170             return
171 
172         msg = "Checking out '{}' with cache '{}'."
173         Logger.info(msg.format(self.to_string(path_info), etag))
174 
175         key = posixpath.join(self.prefix, etag[0:2], etag[2:])
176         from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
177 
178         self._copy(from_info, path_info)
179 
180     def remove(self, path_info):
181         if path_info['scheme'] != 's3':
182             raise NotImplementedError
183 
184         Logger.debug('Removing s3://{}/{}'.format(path_info['bucket'],
185                                                   path_info['key']))
186 
187         self.s3.delete_object(Bucket=path_info['bucket'],
188                               Key=path_info['key'])
189 
190     def md5s_to_path_infos(self, md5s):
191         return [{'scheme': self.scheme,
192                  'bucket': self.bucket,
193                  'key': posixpath.join(self.prefix,
194                                        md5[0:2], md5[2:])} for md5 in md5s]
195 
196     def _all_keys(self):
197         s3 = self.s3
198 
199         keys = []
200         kwargs = {'Bucket': self.bucket,
201                   'Prefix': self.prefix}
202         while True:
203             resp = s3.list_objects_v2(**kwargs)
204             contents = resp.get('Contents', None)
205             if not contents:
206                 break
207 
208             for obj in contents:
209                 keys.append(obj['Key'])
210 
211             token = resp.get('NextContinuationToken', None)
212             if not token:
213                 break
214 
215             kwargs['ContinuationToken'] = token
216 
217         return keys
218 
219     def exists(self, path_infos):
220         # NOTE: We mostly use exists() method when filtering a bulk of cache
221         # files to decide if we need to download/upload them and in s3
222         # list_objects_v2() is much-much faster than trying to check keys
223         # one-by-one.
224         ret = []
225 
226         keys = self._all_keys()
227 
228         for path_info in path_infos:
229             exists = False
230             if path_info['key'] in keys:
231                 exists = True
232             ret.append(exists)
233 
234         return ret
235 
236     def upload(self, from_infos, to_infos, names=None):
237         names = self._verify_path_args(to_infos, from_infos, names)
238 
239         s3 = self.s3
240 
241         for from_info, to_info, name in zip(from_infos, to_infos, names):
242             if to_info['scheme'] != 's3':
243                 raise NotImplementedError
244 
245             if from_info['scheme'] != 'local':
246                 raise NotImplementedError
247 
248             Logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
249                                                             to_info['bucket'],
250                                                             to_info['key']))
251 
252             if not name:
253                 name = os.path.basename(from_info['path'])
254 
255             total = os.path.getsize(from_info['path'])
256             cb = Callback(name, total)
257 
258             try:
259                 s3.upload_file(from_info['path'],
260                                to_info['bucket'],
261                                to_info['key'],
262                                Callback=cb)
263             except Exception as exc:
264                 msg = "Failed to upload '{}'".format(from_info['path'])
265                 Logger.error(msg, exc)
266                 continue
267 
268             progress.finish_target(name)
269 
270     def download(self,
271                  from_infos,
272                  to_infos,
273                  no_progress_bar=False,
274                  names=None):
275         names = self._verify_path_args(from_infos, to_infos, names)
276 
277         s3 = self.s3
278 
279         for to_info, from_info, name in zip(to_infos, from_infos, names):
280             if from_info['scheme'] != 's3':
281                 raise NotImplementedError
282 
283             if to_info['scheme'] == 's3':
284                 self._copy(from_info, to_info, s3=s3)
285                 continue
286 
287             if to_info['scheme'] != 'local':
288                 raise NotImplementedError
289 
290             msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'],
291                                                        from_info['key'],
292                                                        to_info['path'])
293             Logger.debug(msg)
294 
295             tmp_file = self.tmp_file(to_info['path'])
296             if not name:
297                 name = os.path.basename(to_info['path'])
298 
299             if no_progress_bar:
300                 cb = None
301             else:
302                 total = s3.head_object(Bucket=from_info['bucket'],
303                                        Key=from_info['key'])['ContentLength']
304                 cb = Callback(name, total)
305 
306             self._makedirs(to_info['path'])
307 
308             try:
309                 s3.download_file(from_info['bucket'],
310                                  from_info['key'],
311                                  tmp_file,
312                                  Callback=cb)
313             except Exception as exc:
314                 msg = "Failed to download '{}/{}'".format(from_info['bucket'],
315                                                           from_info['key'])
316                 Logger.error(msg, exc)
317                 return
318 
319             os.rename(tmp_file, to_info['path'])
320 
321             if not no_progress_bar:
322                 progress.finish_target(name)
323 
324     def _path_to_etag(self, path):
325         relpath = posixpath.relpath(path, self.prefix)
326         return posixpath.dirname(relpath) + posixpath.basename(relpath)
327 
328     def _all(self):
329         keys = self._all_keys()
330         return [self._path_to_etag(key) for key in keys]
331 
332     def gc(self, cinfos):
333         used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']]
334         used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
335 
336         removed = False
337         for etag in self._all():
338             if etag in used_etags:
339                 continue
340             path_info = {'scheme': 's3',
341                          'key': posixpath.join(self.prefix,
342                                                etag[0:2], etag[2:]),
343                          'bucket': self.bucket}
344             self.remove(path_info)
345             removed = True
346 
347         return removed
348
```
--- 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
@@ -47,7 +47,9 @@
                                            '').lstrip('/')
         self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
         self.region = config.get(Config.SECTION_AWS_REGION, None)
-        self.profile = config.get(Config.SECTION_AWS_PROFILE, 'default')
+        self.profile = os.getenv('AWS_PROFILE',
+                                 config.get(Config.SECTION_AWS_PROFILE,
+                                            'default'))
         self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)
 
         credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)
@@ -62,6 +64,12 @@
         self.aws_access_key_id = creds.get('aws_access_key_id', None)
         self.aws_secret_access_key = creds.get('aws_secret_access_key', None)
 
+        self.region = os.getenv('AWS_DEFAULT_REGION', self.region)
+        self.aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID',
+                                           self.aws_access_key_id)
+        self.aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY',
+                                               self.aws_secret_access_key)
+
     @property
     def bucket(self):
         return urlparse(self.url).netloc

{"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@@ -47,7 +47,9 @@\n                                            '').lstrip('/')\n         self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)\n         self.region = config.get(Config.SECTION_AWS_REGION, None)\n-        self.profile = config.get(Config.SECTION_AWS_PROFILE, 'default')\n+        self.profile = os.getenv('AWS_PROFILE',\n+                                 config.get(Config.SECTION_AWS_PROFILE,\n+                                            'default'))\n         self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)\n \n         credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)\n@@ -62,6 +64,12 @@\n         self.aws_access_key_id = creds.get('aws_access_key_id', None)\n         self.aws_secret_access_key = creds.get('aws_secret_access_key', None)\n \n+        self.region = os.getenv('AWS_DEFAULT_REGION', self.region)\n+        self.aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID',\n+                                           self.aws_access_key_id)\n+        self.aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY',\n+                                               self.aws_secret_access_key)\n+\n     @property\n     def bucket(self):\n         return urlparse(self.url).netloc\n", "issue": "s3: Support AWS credential settings from environment variables\ndvc should also load credential information from system environment variables as awscli does.\r\n\r\nReference: https://docs.aws.amazon.com/cli/latest/userguide/cli-environment.html\r\n\r\nFor example: we should be able to run `dvc pull` despite absence of AWS profiles in _~/.aws_ if we had set the following environment variables:\r\n\r\n```\r\n$ export AWS_ACCESS_KEY_ID=somekey\r\n$ export AWS_SECRET_ACCESS_KEY=somekey\r\n$ export AWS_DEFAULT_REGION=someregion\r\n```\r\n\r\nThe following command works:\r\n``` \r\naws s3 ls s3://bucketname\r\n```\r\n\r\nBut this one does not:\r\n```\r\ndvc pull\r\n```\r\n\r\nA usecase for this is doing `dvc pull` in docker image building process, which would be much easier if if we can override configuration via environment variables. \n", "before_files": [{"content": "import os\nimport threading\nimport posixpath\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\ntry:\n    from urlparse import urlparse\nexcept ImportError:\n    from urllib.parse import urlparse\n\nfrom dvc.logger import Logger\nfrom dvc.progress import progress\nfrom dvc.config import Config\nfrom dvc.remote.base import RemoteBase\nfrom dvc.remote.local import RemoteLOCAL\nfrom dvc.exceptions import DvcException\n\n\nclass Callback(object):\n    def __init__(self, name, total):\n        self.name = name\n        self.total = total\n        self.current = 0\n        self.lock = threading.Lock()\n\n    def __call__(self, byts):\n        with self.lock:\n            self.current += byts\n            progress.update_target(self.name, self.current, self.total)\n\n\nclass RemoteS3(RemoteBase):\n    scheme = 's3'\n    REGEX = r'^s3://(?P<path>.*)$'\n    REQUIRES = {'boto3': boto3}\n    PARAM_ETAG = 'etag'\n\n    def __init__(self, project, config):\n        import configobj\n\n        self.project = project\n        storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH,\n                                           '').lstrip('/')\n        self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)\n        self.region = config.get(Config.SECTION_AWS_REGION, None)\n        self.profile = config.get(Config.SECTION_AWS_PROFILE, 'default')\n        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)\n\n        credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)\n        if credentialpath:\n            creds_conf = configobj.ConfigObj(credentialpath)\n            creds = creds_conf.get(self.profile, {})\n        else:\n            creds = {}\n\n        self.creds = creds\n        self.region = creds.get('region', self.region)\n        self.aws_access_key_id = creds.get('aws_access_key_id', None)\n        self.aws_secret_access_key = creds.get('aws_secret_access_key', None)\n\n    @property\n    def bucket(self):\n        return urlparse(self.url).netloc\n\n    @property\n    def prefix(self):\n        return urlparse(self.url).path.lstrip('/')\n\n    @property\n    def s3(self):\n        if not self.creds:\n            session = boto3.session.Session(profile_name=self.profile)\n        else:\n            session = boto3.session.Session(\n                             aws_access_key_id=self.aws_access_key_id,\n                             aws_secret_access_key=self.aws_secret_access_key,\n                             region_name=self.region)\n        return session.client('s3', endpoint_url=self.endpoint_url)\n\n    def get_etag(self, bucket, key):\n        try:\n            obj = self.s3.head_object(Bucket=bucket, Key=key)\n        except Exception:\n            raise DvcException('s3://{}/{} does not exist'.format(bucket, key))\n\n        return obj['ETag'].strip('\"')\n\n    def save_info(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        return {self.PARAM_ETAG: self.get_etag(path_info['bucket'],\n                                               path_info['key'])}\n\n    def changed(self, path_info, checksum_info):\n        if not self.exists([path_info])[0]:\n            return True\n\n        etag = checksum_info.get(self.PARAM_ETAG, None)\n        if etag is None:\n            return True\n\n        if self.changed_cache(etag):\n            return True\n\n        return checksum_info != self.save_info(path_info)\n\n    def _copy(self, from_info, to_info, s3=None):\n        s3 = s3 if s3 else self.s3\n\n        source = {'Bucket': from_info['bucket'],\n                  'Key': from_info['key']}\n        self.s3.copy(source, to_info['bucket'], to_info['key'])\n\n    def save(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        etag = self.get_etag(path_info['bucket'], path_info['key'])\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        self._copy(path_info, to_info)\n\n        return {self.PARAM_ETAG: etag}\n\n    @staticmethod\n    def to_string(path_info):\n        return \"s3://{}/{}\".format(path_info['bucket'], path_info['key'])\n\n    def changed_cache(self, etag):\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        if {self.PARAM_ETAG: etag} != self.save_info(cache):\n            if self.exists([cache])[0]:\n                msg = 'Corrupted cache file {}'\n                Logger.warn(msg.format(self.to_string(cache)))\n                self.remove(cache)\n            return True\n\n        return False\n\n    def checkout(self, path_info, checksum_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        etag = checksum_info.get(self.PARAM_ETAG, None)\n        if not etag:\n            return\n\n        if not self.changed(path_info, checksum_info):\n            msg = \"Data '{}' didn't change.\"\n            Logger.info(msg.format(self.to_string(path_info)))\n            return\n\n        if self.changed_cache(etag):\n            msg = \"Cache '{}' not found. File '{}' won't be created.\"\n            Logger.warn(msg.format(etag, self.to_string(path_info)))\n            return\n\n        if self.exists([path_info])[0]:\n            msg = \"Data '{}' exists. Removing before checkout.\"\n            Logger.warn(msg.format(self.to_string(path_info)))\n            self.remove(path_info)\n            return\n\n        msg = \"Checking out '{}' with cache '{}'.\"\n        Logger.info(msg.format(self.to_string(path_info), etag))\n\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        self._copy(from_info, path_info)\n\n    def remove(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        Logger.debug('Removing s3://{}/{}'.format(path_info['bucket'],\n                                                  path_info['key']))\n\n        self.s3.delete_object(Bucket=path_info['bucket'],\n                              Key=path_info['key'])\n\n    def md5s_to_path_infos(self, md5s):\n        return [{'scheme': self.scheme,\n                 'bucket': self.bucket,\n                 'key': posixpath.join(self.prefix,\n                                       md5[0:2], md5[2:])} for md5 in md5s]\n\n    def _all_keys(self):\n        s3 = self.s3\n\n        keys = []\n        kwargs = {'Bucket': self.bucket,\n                  'Prefix': self.prefix}\n        while True:\n            resp = s3.list_objects_v2(**kwargs)\n            contents = resp.get('Contents', None)\n            if not contents:\n                break\n\n            for obj in contents:\n                keys.append(obj['Key'])\n\n            token = resp.get('NextContinuationToken', None)\n            if not token:\n                break\n\n            kwargs['ContinuationToken'] = token\n\n        return keys\n\n    def exists(self, path_infos):\n        # NOTE: We mostly use exists() method when filtering a bulk of cache\n        # files to decide if we need to download/upload them and in s3\n        # list_objects_v2() is much-much faster than trying to check keys\n        # one-by-one.\n        ret = []\n\n        keys = self._all_keys()\n\n        for path_info in path_infos:\n            exists = False\n            if path_info['key'] in keys:\n                exists = True\n            ret.append(exists)\n\n        return ret\n\n    def upload(self, from_infos, to_infos, names=None):\n        names = self._verify_path_args(to_infos, from_infos, names)\n\n        s3 = self.s3\n\n        for from_info, to_info, name in zip(from_infos, to_infos, names):\n            if to_info['scheme'] != 's3':\n                raise NotImplementedError\n\n            if from_info['scheme'] != 'local':\n                raise NotImplementedError\n\n            Logger.debug(\"Uploading '{}' to '{}/{}'\".format(from_info['path'],\n                                                            to_info['bucket'],\n                                                            to_info['key']))\n\n            if not name:\n                name = os.path.basename(from_info['path'])\n\n            total = os.path.getsize(from_info['path'])\n            cb = Callback(name, total)\n\n            try:\n                s3.upload_file(from_info['path'],\n                               to_info['bucket'],\n                               to_info['key'],\n                               Callback=cb)\n            except Exception as exc:\n                msg = \"Failed to upload '{}'\".format(from_info['path'])\n                Logger.error(msg, exc)\n                continue\n\n            progress.finish_target(name)\n\n    def download(self,\n                 from_infos,\n                 to_infos,\n                 no_progress_bar=False,\n                 names=None):\n        names = self._verify_path_args(from_infos, to_infos, names)\n\n        s3 = self.s3\n\n        for to_info, from_info, name in zip(to_infos, from_infos, names):\n            if from_info['scheme'] != 's3':\n                raise NotImplementedError\n\n            if to_info['scheme'] == 's3':\n                self._copy(from_info, to_info, s3=s3)\n                continue\n\n            if to_info['scheme'] != 'local':\n                raise NotImplementedError\n\n            msg = \"Downloading '{}/{}' to '{}'\".format(from_info['bucket'],\n                                                       from_info['key'],\n                                                       to_info['path'])\n            Logger.debug(msg)\n\n            tmp_file = self.tmp_file(to_info['path'])\n            if not name:\n                name = os.path.basename(to_info['path'])\n\n            if no_progress_bar:\n                cb = None\n            else:\n                total = s3.head_object(Bucket=from_info['bucket'],\n                                       Key=from_info['key'])['ContentLength']\n                cb = Callback(name, total)\n\n            self._makedirs(to_info['path'])\n\n            try:\n                s3.download_file(from_info['bucket'],\n                                 from_info['key'],\n                                 tmp_file,\n                                 Callback=cb)\n            except Exception as exc:\n                msg = \"Failed to download '{}/{}'\".format(from_info['bucket'],\n                                                          from_info['key'])\n                Logger.error(msg, exc)\n                return\n\n            os.rename(tmp_file, to_info['path'])\n\n            if not no_progress_bar:\n                progress.finish_target(name)\n\n    def _path_to_etag(self, path):\n        relpath = posixpath.relpath(path, self.prefix)\n        return posixpath.dirname(relpath) + posixpath.basename(relpath)\n\n    def _all(self):\n        keys = self._all_keys()\n        return [self._path_to_etag(key) for key in keys]\n\n    def gc(self, cinfos):\n        used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']]\n        used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]\n\n        removed = False\n        for etag in self._all():\n            if etag in used_etags:\n                continue\n            path_info = {'scheme': 's3',\n                         'key': posixpath.join(self.prefix,\n                                               etag[0:2], etag[2:]),\n                         'bucket': self.bucket}\n            self.remove(path_info)\n            removed = True\n\n        return removed\n", "path": "dvc/remote/s3.py"}], "after_files": [{"content": "import os\nimport threading\nimport posixpath\n\ntry:\n    import boto3\nexcept ImportError:\n    boto3 = None\n\ntry:\n    from urlparse import urlparse\nexcept ImportError:\n    from urllib.parse import urlparse\n\nfrom dvc.logger import Logger\nfrom dvc.progress import progress\nfrom dvc.config import Config\nfrom dvc.remote.base import RemoteBase\nfrom dvc.remote.local import RemoteLOCAL\nfrom dvc.exceptions import DvcException\n\n\nclass Callback(object):\n    def __init__(self, name, total):\n        self.name = name\n        self.total = total\n        self.current = 0\n        self.lock = threading.Lock()\n\n    def __call__(self, byts):\n        with self.lock:\n            self.current += byts\n            progress.update_target(self.name, self.current, self.total)\n\n\nclass RemoteS3(RemoteBase):\n    scheme = 's3'\n    REGEX = r'^s3://(?P<path>.*)$'\n    REQUIRES = {'boto3': boto3}\n    PARAM_ETAG = 'etag'\n\n    def __init__(self, project, config):\n        import configobj\n\n        self.project = project\n        storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH,\n                                           '').lstrip('/')\n        self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)\n        self.region = config.get(Config.SECTION_AWS_REGION, None)\n        self.profile = os.getenv('AWS_PROFILE',\n                                 config.get(Config.SECTION_AWS_PROFILE,\n                                            'default'))\n        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)\n\n        credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)\n        if credentialpath:\n            creds_conf = configobj.ConfigObj(credentialpath)\n            creds = creds_conf.get(self.profile, {})\n        else:\n            creds = {}\n\n        self.creds = creds\n        self.region = creds.get('region', self.region)\n        self.aws_access_key_id = creds.get('aws_access_key_id', None)\n        self.aws_secret_access_key = creds.get('aws_secret_access_key', None)\n\n        self.region = os.getenv('AWS_DEFAULT_REGION', self.region)\n        self.aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID',\n                                           self.aws_access_key_id)\n        self.aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY',\n                                               self.aws_secret_access_key)\n\n    @property\n    def bucket(self):\n        return urlparse(self.url).netloc\n\n    @property\n    def prefix(self):\n        return urlparse(self.url).path.lstrip('/')\n\n    @property\n    def s3(self):\n        if not self.creds:\n            session = boto3.session.Session(profile_name=self.profile)\n        else:\n            session = boto3.session.Session(\n                             aws_access_key_id=self.aws_access_key_id,\n                             aws_secret_access_key=self.aws_secret_access_key,\n                             region_name=self.region)\n        return session.client('s3', endpoint_url=self.endpoint_url)\n\n    def get_etag(self, bucket, key):\n        try:\n            obj = self.s3.head_object(Bucket=bucket, Key=key)\n        except Exception:\n            raise DvcException('s3://{}/{} does not exist'.format(bucket, key))\n\n        return obj['ETag'].strip('\"')\n\n    def save_info(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        return {self.PARAM_ETAG: self.get_etag(path_info['bucket'],\n                                               path_info['key'])}\n\n    def changed(self, path_info, checksum_info):\n        if not self.exists([path_info])[0]:\n            return True\n\n        etag = checksum_info.get(self.PARAM_ETAG, None)\n        if etag is None:\n            return True\n\n        if self.changed_cache(etag):\n            return True\n\n        return checksum_info != self.save_info(path_info)\n\n    def _copy(self, from_info, to_info, s3=None):\n        s3 = s3 if s3 else self.s3\n\n        source = {'Bucket': from_info['bucket'],\n                  'Key': from_info['key']}\n        self.s3.copy(source, to_info['bucket'], to_info['key'])\n\n    def save(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        etag = self.get_etag(path_info['bucket'], path_info['key'])\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        self._copy(path_info, to_info)\n\n        return {self.PARAM_ETAG: etag}\n\n    @staticmethod\n    def to_string(path_info):\n        return \"s3://{}/{}\".format(path_info['bucket'], path_info['key'])\n\n    def changed_cache(self, etag):\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        if {self.PARAM_ETAG: etag} != self.save_info(cache):\n            if self.exists([cache])[0]:\n                msg = 'Corrupted cache file {}'\n                Logger.warn(msg.format(self.to_string(cache)))\n                self.remove(cache)\n            return True\n\n        return False\n\n    def checkout(self, path_info, checksum_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        etag = checksum_info.get(self.PARAM_ETAG, None)\n        if not etag:\n            return\n\n        if not self.changed(path_info, checksum_info):\n            msg = \"Data '{}' didn't change.\"\n            Logger.info(msg.format(self.to_string(path_info)))\n            return\n\n        if self.changed_cache(etag):\n            msg = \"Cache '{}' not found. File '{}' won't be created.\"\n            Logger.warn(msg.format(etag, self.to_string(path_info)))\n            return\n\n        if self.exists([path_info])[0]:\n            msg = \"Data '{}' exists. Removing before checkout.\"\n            Logger.warn(msg.format(self.to_string(path_info)))\n            self.remove(path_info)\n            return\n\n        msg = \"Checking out '{}' with cache '{}'.\"\n        Logger.info(msg.format(self.to_string(path_info), etag))\n\n        key = posixpath.join(self.prefix, etag[0:2], etag[2:])\n        from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}\n\n        self._copy(from_info, path_info)\n\n    def remove(self, path_info):\n        if path_info['scheme'] != 's3':\n            raise NotImplementedError\n\n        Logger.debug('Removing s3://{}/{}'.format(path_info['bucket'],\n                                                  path_info['key']))\n\n        self.s3.delete_object(Bucket=path_info['bucket'],\n                              Key=path_info['key'])\n\n    def md5s_to_path_infos(self, md5s):\n        return [{'scheme': self.scheme,\n                 'bucket': self.bucket,\n                 'key': posixpath.join(self.prefix,\n                                       md5[0:2], md5[2:])} for md5 in md5s]\n\n    def _all_keys(self):\n        s3 = self.s3\n\n        keys = []\n        kwargs = {'Bucket': self.bucket,\n                  'Prefix': self.prefix}\n        while True:\n            resp = s3.list_objects_v2(**kwargs)\n            contents = resp.get('Contents', None)\n            if not contents:\n                break\n\n            for obj in contents:\n                keys.append(obj['Key'])\n\n            token = resp.get('NextContinuationToken', None)\n            if not token:\n                break\n\n            kwargs['ContinuationToken'] = token\n\n        return keys\n\n    def exists(self, path_infos):\n        # NOTE: We mostly use exists() method when filtering a bulk of cache\n        # files to decide if we need to download/upload them and in s3\n        # list_objects_v2() is much-much faster than trying to check keys\n        # one-by-one.\n        ret = []\n\n        keys = self._all_keys()\n\n        for path_info in path_infos:\n            exists = False\n            if path_info['key'] in keys:\n                exists = True\n            ret.append(exists)\n\n        return ret\n\n    def upload(self, from_infos, to_infos, names=None):\n        names = self._verify_path_args(to_infos, from_infos, names)\n\n        s3 = self.s3\n\n        for from_info, to_info, name in zip(from_infos, to_infos, names):\n            if to_info['scheme'] != 's3':\n                raise NotImplementedError\n\n            if from_info['scheme'] != 'local':\n                raise NotImplementedError\n\n            Logger.debug(\"Uploading '{}' to '{}/{}'\".format(from_info['path'],\n                                                            to_info['bucket'],\n                                                            to_info['key']))\n\n            if not name:\n                name = os.path.basename(from_info['path'])\n\n            total = os.path.getsize(from_info['path'])\n            cb = Callback(name, total)\n\n            try:\n                s3.upload_file(from_info['path'],\n                               to_info['bucket'],\n                               to_info['key'],\n                               Callback=cb)\n            except Exception as exc:\n                msg = \"Failed to upload '{}'\".format(from_info['path'])\n                Logger.error(msg, exc)\n                continue\n\n            progress.finish_target(name)\n\n    def download(self,\n                 from_infos,\n                 to_infos,\n                 no_progress_bar=False,\n                 names=None):\n        names = self._verify_path_args(from_infos, to_infos, names)\n\n        s3 = self.s3\n\n        for to_info, from_info, name in zip(to_infos, from_infos, names):\n            if from_info['scheme'] != 's3':\n                raise NotImplementedError\n\n            if to_info['scheme'] == 's3':\n                self._copy(from_info, to_info, s3=s3)\n                continue\n\n            if to_info['scheme'] != 'local':\n                raise NotImplementedError\n\n            msg = \"Downloading '{}/{}' to '{}'\".format(from_info['bucket'],\n                                                       from_info['key'],\n                                                       to_info['path'])\n            Logger.debug(msg)\n\n            tmp_file = self.tmp_file(to_info['path'])\n            if not name:\n                name = os.path.basename(to_info['path'])\n\n            if no_progress_bar:\n                cb = None\n            else:\n                total = s3.head_object(Bucket=from_info['bucket'],\n                                       Key=from_info['key'])['ContentLength']\n                cb = Callback(name, total)\n\n            self._makedirs(to_info['path'])\n\n            try:\n                s3.download_file(from_info['bucket'],\n                                 from_info['key'],\n                                 tmp_file,\n                                 Callback=cb)\n            except Exception as exc:\n                msg = \"Failed to download '{}/{}'\".format(from_info['bucket'],\n                                                          from_info['key'])\n                Logger.error(msg, exc)\n                return\n\n            os.rename(tmp_file, to_info['path'])\n\n            if not no_progress_bar:\n                progress.finish_target(name)\n\n    def _path_to_etag(self, path):\n        relpath = posixpath.relpath(path, self.prefix)\n        return posixpath.dirname(relpath) + posixpath.basename(relpath)\n\n    def _all(self):\n        keys = self._all_keys()\n        return [self._path_to_etag(key) for key in keys]\n\n    def gc(self, cinfos):\n        used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']]\n        used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]\n\n        removed = False\n        for etag in self._all():\n            if etag in used_etags:\n                continue\n            path_info = {'scheme': 's3',\n                         'key': posixpath.join(self.prefix,\n                                               etag[0:2], etag[2:]),\n                         'bucket': self.bucket}\n            self.remove(path_info)\n            removed = True\n\n        return removed\n", "path": "dvc/remote/s3.py"}]}