We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
S3 AWS: default profile is always used
Hi!

Defining a profile in dvc-config has no influence on the used profile in AWS credentials file.

I quickly took a look at the code and  I never see boto3.session(profile=..) called. Or is this done using the AWS_PROFILE env variable?

(using dvc 0.14.0)

--- 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 math
3 import threading
4 import posixpath
5 
6 try:
7     import boto3
8 except ImportError:
9     boto3 = None
10 
11 try:
12     from urlparse import urlparse
13 except ImportError:
14     from urllib.parse import urlparse
15 
16 from dvc.logger import Logger
17 from dvc.progress import progress
18 from dvc.config import Config
19 from dvc.remote.base import RemoteBase
20 from dvc.remote.local import RemoteLOCAL
21 from dvc.exceptions import DvcException
22 
23 
24 class Callback(object):
25     def __init__(self, name, total):
26         self.name = name
27         self.total = total
28         self.current = 0
29         self.lock = threading.Lock()
30 
31     def __call__(self, byts):
32         with self.lock:
33             self.current += byts
34             progress.update_target(self.name, self.current, self.total)
35 
36 
37 class RemoteS3(RemoteBase):
38     scheme = 's3'
39     REGEX = r'^s3://(?P<path>.*)$'
40     REQUIRES = {'boto3': boto3}
41     PARAM_ETAG = 'etag'
42 
43     def __init__(self, project, config):
44         self.project = project
45         storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH, '').lstrip('/')
46         self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
47         self.region = config.get(Config.SECTION_AWS_REGION, None)
48         self.profile = config.get(Config.SECTION_AWS_PROFILE, None)
49         self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)
50         self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, None)
51 
52     @property
53     def bucket(self):
54         return urlparse(self.url).netloc
55 
56     @property
57     def prefix(self):
58         return urlparse(self.url).path.lstrip('/')
59 
60     @property
61     def s3(self):
62         return boto3.resource('s3', endpoint_url=self.endpoint_url)
63 
64     @property
65     def s3_session_client(self):
66         session = boto3.session.Session()
67         s3 = session.client('s3', endpoint_url=self.endpoint_url)
68         return s3
69 
70     def get_etag(self, bucket, key):
71         try:
72             obj = self.s3.Object(bucket, key).get()
73         except Exception:
74             raise DvcException('s3://{}/{} does not exist'.format(bucket, key))
75 
76         return obj['ETag'].strip('"')
77 
78     def save_info(self, path_info):
79         if path_info['scheme'] != 's3':
80             raise NotImplementedError
81 
82         return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])}
83 
84     def _copy(self, from_info, to_info, s3=None):
85         s3 = s3 if s3 else self.s3
86 
87         source = {'Bucket': from_info['bucket'],
88                   'Key': from_info['key']}
89         self.s3.Bucket(to_info['bucket']).copy(source, to_info['key'])
90 
91     def save(self, path_info):
92         if path_info['scheme'] != 's3':
93             raise NotImplementedError
94 
95         etag = self.get_etag(path_info['bucket'], path_info['key'])
96         key = posixpath.join(self.prefix, etag[0:2], etag[2:])
97         to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
98 
99         self._copy(path_info, to_info)
100 
101         return {self.PARAM_ETAG: etag}
102 
103     def checkout(self, path_info, checksum_info):
104         if path_info['scheme'] != 's3':
105             raise NotImplementedError
106 
107         etag = checksum_info.get(self.PARAM_ETAG, None)
108         if not etag:
109             return
110 
111         key = posixpath.join(self.prefix, etag[0:2], etag[2:])
112         from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
113 
114         self._copy(from_info, path_info)
115 
116     def remove(self, path_info):
117         if path_info['scheme'] != 's3':
118             raise NotImplementedError
119 
120         Logger.debug('Removing s3://{}/{}'.format(path_info['bucket'],
121                                                   path_info['key']))
122 
123         obj = self.s3.Object(path_info['bucket'], path_info['key'])
124         try:
125             obj.get()
126         except Exception:
127             return
128 
129         obj.delete()
130 
131     def md5s_to_path_infos(self, md5s):
132         return [{'scheme': self.scheme,
133                  'bucket': self.bucket,
134                  'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]
135 
136     def exists(self, path_infos):
137         # NOTE: We mostly use exists() method when filtering a bulk of cache
138         # files to decide if we need to download/upload them and in s3
139         # list_objects_v2() is much-much faster than trying to check keys
140         # one-by-one.
141         ret = []
142         s3 = self.s3_session_client
143 
144         keys = []
145         kwargs = {'Bucket': self.bucket,
146                   'Prefix': self.prefix}
147         while True:
148             resp = s3.list_objects_v2(**kwargs)
149             contents = resp.get('Contents', None)
150             if not contents:
151                 break
152 
153             for obj in contents:
154                 keys.append(obj['Key'])
155 
156             token = resp.get('NextContinuationToken', None)
157             if not token:
158                 break
159 
160             kwargs['ContinuationToken'] = token
161 
162         for path_info in path_infos:
163             exists = False
164             if path_info['key'] in keys:
165                 exists = True
166             ret.append(exists)
167 
168         return ret
169 
170     def upload(self, from_infos, to_infos, names=None):
171         names = self._verify_path_args(to_infos, from_infos, names)
172 
173         s3 = self.s3_session_client
174 
175         for from_info, to_info, name in zip(from_infos, to_infos, names):
176             if to_info['scheme'] != 's3':
177                 raise NotImplementedError
178 
179             if from_info['scheme'] != 'local':
180                 raise NotImplementedError
181 
182             Logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
183                                                             to_info['bucket'],
184                                                             to_info['key']))
185 
186             if not name:
187                 name = os.path.basename(from_info['path'])
188 
189             total = os.path.getsize(from_info['path'])
190             cb = Callback(name, total)
191 
192             try:
193                 s3.upload_file(from_info['path'], to_info['bucket'], to_info['key'], Callback=cb)
194             except Exception as exc:
195                 Logger.error("Failed to upload '{}'".format(from_info['path']), exc)
196                 continue
197 
198             progress.finish_target(name)
199 
200     def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
201         names = self._verify_path_args(from_infos, to_infos, names)
202 
203         s3 = self.s3_session_client
204 
205         for to_info, from_info, name in zip(to_infos, from_infos, names):
206             if from_info['scheme'] != 's3':
207                 raise NotImplementedError
208 
209             if to_info['scheme'] == 's3':
210                 self._copy(from_info, to_info, s3=s3)
211                 continue
212 
213             if to_info['scheme'] != 'local':
214                 raise NotImplementedError
215 
216             Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['bucket'],
217                                                               from_info['key'],
218                                                               to_info['path']))
219 
220             tmp_file = self.tmp_file(to_info['path'])
221             if not name:
222                 name = os.path.basename(to_info['path'])
223 
224             if no_progress_bar:
225                 cb = None
226             else:
227                 total = s3.head_object(Bucket=from_info['bucket'],
228                                        Key=from_info['key'])['ContentLength']
229                 cb = Callback(name, total)
230 
231             self._makedirs(to_info['path'])
232 
233             try:
234                 s3.download_file(from_info['bucket'], from_info['key'], tmp_file, Callback=cb)
235             except Exception as exc:
236                 Logger.error("Failed to download '{}/{}'".format(from_info['bucket'],
237                                                                  from_info['key']), exc)
238                 return
239 
240             os.rename(tmp_file, to_info['path'])
241 
242             if not no_progress_bar:
243                 progress.finish_target(name)
244 
245     def _path_to_etag(self, path):
246         relpath = posixpath.relpath(path, self.prefix)
247         return posixpath.dirname(relpath) + posixpath.basename(relpath)
248 
249     def _all(self):
250         objects = self.s3.Bucket(self.bucket).objects.filter(Prefix=self.prefix)
251         return [self._path_to_etag(obj.key) for obj in objects]
252 
253     def gc(self, checksum_infos):
254         used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['s3']]
255         used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']]
256 
257         for etag in self._all():
258             if etag in used_etags:
259                 continue
260             path_info = {'scheme': 's3',
261                          'key': posixpath.join(self.prefix, etag[0:2], etag[2:]),
262                          'bucket': self.bucket}
263             self.remove(path_info)
264
```
--- 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
@@ -41,14 +41,26 @@
     PARAM_ETAG = 'etag'
 
     def __init__(self, project, config):
+        import configobj
+
         self.project = project
         storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH, '').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, None)
-        self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)
+        self.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)
+        if credentialpath:
+            creds_conf = configobj.ConfigObj(credentialpath)
+            creds = creds_conf.get(self.profile, {})
+        else:
+            creds = {}
+
+        self.region = creds.get('region', self.region)
+        self.aws_access_key_id = creds.get('aws_access_key_id', None)
+        self.aws_secret_access_key = creds.get('aws_secret_access_key', None)
+
     @property
     def bucket(self):
         return urlparse(self.url).netloc
@@ -59,13 +71,12 @@
 
     @property
     def s3(self):
-        return boto3.resource('s3', endpoint_url=self.endpoint_url)
-
-    @property
-    def s3_session_client(self):
-        session = boto3.session.Session()
-        s3 = session.client('s3', endpoint_url=self.endpoint_url)
-        return s3
+        session = boto3.session.Session(profile_name=self.profile)
+        return session.client('s3',
+                              aws_access_key_id=self.aws_access_key_id,
+                              aws_secret_access_key=self.aws_secret_access_key,
+                              region_name=self.region,
+                              endpoint_url=self.endpoint_url)
 
     def get_etag(self, bucket, key):
         try:
@@ -139,7 +150,7 @@
         # list_objects_v2() is much-much faster than trying to check keys
         # one-by-one.
         ret = []
-        s3 = self.s3_session_client
+        s3 = self.s3
 
         keys = []
         kwargs = {'Bucket': self.bucket,
@@ -170,7 +181,7 @@
     def upload(self, from_infos, to_infos, names=None):
         names = self._verify_path_args(to_infos, from_infos, names)
 
-        s3 = self.s3_session_client
+        s3 = self.s3
 
         for from_info, to_info, name in zip(from_infos, to_infos, names):
             if to_info['scheme'] != 's3':
@@ -200,7 +211,7 @@
     def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
         names = self._verify_path_args(from_infos, to_infos, names)
 
-        s3 = self.s3_session_client
+        s3 = self.s3
 
         for to_info, from_info, name in zip(to_infos, from_infos, names):
             if from_info['scheme'] != 's3':

{"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@@ -41,14 +41,26 @@\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, '').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, None)\n-        self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, 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.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@@ -59,13 +71,12 @@\n \n     @property\n     def s3(self):\n-        return boto3.resource('s3', endpoint_url=self.endpoint_url)\n-\n-    @property\n-    def s3_session_client(self):\n-        session = boto3.session.Session()\n-        s3 = session.client('s3', endpoint_url=self.endpoint_url)\n-        return s3\n+        session = boto3.session.Session(profile_name=self.profile)\n+        return session.client('s3',\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+                              endpoint_url=self.endpoint_url)\n \n     def get_etag(self, bucket, key):\n         try:\n@@ -139,7 +150,7 @@\n         # list_objects_v2() is much-much faster than trying to check keys\n         # one-by-one.\n         ret = []\n-        s3 = self.s3_session_client\n+        s3 = self.s3\n \n         keys = []\n         kwargs = {'Bucket': self.bucket,\n@@ -170,7 +181,7 @@\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_session_client\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@@ -200,7 +211,7 @@\n     def download(self, from_infos, to_infos, no_progress_bar=False, names=None):\n         names = self._verify_path_args(from_infos, to_infos, names)\n \n-        s3 = self.s3_session_client\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", "issue": "S3 AWS: default profile is always used\nHi!\r\n\r\nDefining a profile in dvc-config has no influence on the used profile in AWS credentials file.\r\n\r\nI quickly took a look at the code and  I never see boto3.session(profile=..) called. Or is this done using the AWS_PROFILE env variable?\r\n\r\n(using dvc 0.14.0)\n", "before_files": [{"content": "import os\nimport math\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        self.project = project\n        storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH, '').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, None)\n        self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None)\n        self.endpoint_url = config.get(Config.SECTION_AWS_ENDPOINT_URL, 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        return boto3.resource('s3', endpoint_url=self.endpoint_url)\n\n    @property\n    def s3_session_client(self):\n        session = boto3.session.Session()\n        s3 = session.client('s3', endpoint_url=self.endpoint_url)\n        return s3\n\n    def get_etag(self, bucket, key):\n        try:\n            obj = self.s3.Object(bucket, key).get()\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'], path_info['key'])}\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.Bucket(to_info['bucket']).copy(source, 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    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        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        obj = self.s3.Object(path_info['bucket'], path_info['key'])\n        try:\n            obj.get()\n        except Exception:\n            return\n\n        obj.delete()\n\n    def md5s_to_path_infos(self, md5s):\n        return [{'scheme': self.scheme,\n                 'bucket': self.bucket,\n                 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]\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        s3 = self.s3_session_client\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        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_session_client\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'], to_info['bucket'], to_info['key'], Callback=cb)\n            except Exception as exc:\n                Logger.error(\"Failed to upload '{}'\".format(from_info['path']), exc)\n                continue\n\n            progress.finish_target(name)\n\n    def download(self, from_infos, to_infos, no_progress_bar=False, names=None):\n        names = self._verify_path_args(from_infos, to_infos, names)\n\n        s3 = self.s3_session_client\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            Logger.debug(\"Downloading '{}/{}' to '{}'\".format(from_info['bucket'],\n                                                              from_info['key'],\n                                                              to_info['path']))\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'], from_info['key'], tmp_file, Callback=cb)\n            except Exception as exc:\n                Logger.error(\"Failed to download '{}/{}'\".format(from_info['bucket'],\n                                                                 from_info['key']), 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        objects = self.s3.Bucket(self.bucket).objects.filter(Prefix=self.prefix)\n        return [self._path_to_etag(obj.key) for obj in objects]\n\n    def gc(self, checksum_infos):\n        used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['s3']]\n        used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']]\n\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, etag[0:2], etag[2:]),\n                         'bucket': self.bucket}\n            self.remove(path_info)\n", "path": "dvc/remote/s3.py"}], "after_files": [{"content": "import os\nimport math\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, '').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.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        session = boto3.session.Session(profile_name=self.profile)\n        return session.client('s3',\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                              endpoint_url=self.endpoint_url)\n\n    def get_etag(self, bucket, key):\n        try:\n            obj = self.s3.Object(bucket, key).get()\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'], path_info['key'])}\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.Bucket(to_info['bucket']).copy(source, 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    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        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        obj = self.s3.Object(path_info['bucket'], path_info['key'])\n        try:\n            obj.get()\n        except Exception:\n            return\n\n        obj.delete()\n\n    def md5s_to_path_infos(self, md5s):\n        return [{'scheme': self.scheme,\n                 'bucket': self.bucket,\n                 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]\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        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        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'], to_info['bucket'], to_info['key'], Callback=cb)\n            except Exception as exc:\n                Logger.error(\"Failed to upload '{}'\".format(from_info['path']), exc)\n                continue\n\n            progress.finish_target(name)\n\n    def download(self, from_infos, to_infos, no_progress_bar=False, 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            Logger.debug(\"Downloading '{}/{}' to '{}'\".format(from_info['bucket'],\n                                                              from_info['key'],\n                                                              to_info['path']))\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'], from_info['key'], tmp_file, Callback=cb)\n            except Exception as exc:\n                Logger.error(\"Failed to download '{}/{}'\".format(from_info['bucket'],\n                                                                 from_info['key']), 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        objects = self.s3.Bucket(self.bucket).objects.filter(Prefix=self.prefix)\n        return [self._path_to_etag(obj.key) for obj in objects]\n\n    def gc(self, checksum_infos):\n        used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['s3']]\n        used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']]\n\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, etag[0:2], etag[2:]),\n                         'bucket': self.bucket}\n            self.remove(path_info)\n", "path": "dvc/remote/s3.py"}]}