We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
s3 sync from local to s3 bucket re-uploads files with version 1.3.6
It is probably related to #718 but it is still happening on my OSX Mavericks with Python 2.7. I sync local files to S3 using "aws s3 sync" but some files always want to re-upload. I did a debug and comparator thinks that "file does not exist at destination".

I even removed all scripts and python packages for awscli, botocore and jmespath, and then reinstalled latest 1.3.6 but no luck. I use awscli for s3 sync since version 1.0 without any issues and have only started to experiences such issues starting with 1.3.2.

Appreciate if you could look into it. Let me know if additional information was needed.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `awscli/customizations/s3/utils.py`
Content:
```
1 # Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License"). You
4 # may not use this file except in compliance with the License. A copy of
5 # the License is located at
6 #
7 #     http://aws.amazon.com/apache2.0/
8 #
9 # or in the "license" file accompanying this file. This file is
10 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11 # ANY KIND, either express or implied. See the License for the specific
12 # language governing permissions and limitations under the License.
13 from datetime import datetime
14 import mimetypes
15 import hashlib
16 import math
17 import os
18 import sys
19 from collections import namedtuple, deque
20 from functools import partial
21 
22 from six import PY3
23 from six.moves import queue
24 from dateutil.parser import parse
25 from dateutil.tz import tzlocal
26 from botocore.compat import unquote_str
27 
28 from awscli.customizations.s3.constants import MAX_PARTS
29 from awscli.customizations.s3.constants import MAX_SINGLE_UPLOAD_SIZE
30 
31 
32 class MD5Error(Exception):
33     """
34     Exception for md5's that do not match.
35     """
36     pass
37 
38 
39 class StablePriorityQueue(queue.Queue):
40     """Priority queue that maintains FIFO order for same priority items.
41 
42     This class was written to handle the tasks created in
43     awscli.customizations.s3.tasks, but it's possible to use this
44     class outside of that context.  In order for this to be the case,
45     the following conditions should be met:
46 
47         * Objects that are queued should have a PRIORITY attribute.
48           This should be an integer value not to exceed the max_priority
49           value passed into the ``__init__``.  Objects with lower
50           priority numbers are retrieved before objects with higher
51           priority numbers.
52         * A relatively small max_priority should be chosen.  ``get()``
53           calls are O(max_priority).
54 
55     Any object that does not have a ``PRIORITY`` attribute or whose
56     priority exceeds ``max_priority`` will be queued at the highest
57     (least important) priority available.
58 
59     """
60     def __init__(self, maxsize=0, max_priority=20):
61         queue.Queue.__init__(self, maxsize=maxsize)
62         self.priorities = [deque([]) for i in range(max_priority + 1)]
63         self.default_priority = max_priority
64 
65     def _qsize(self):
66         size = 0
67         for bucket in self.priorities:
68             size += len(bucket)
69         return size
70 
71     def _put(self, item):
72         priority = min(getattr(item, 'PRIORITY', self.default_priority),
73                         self.default_priority)
74         self.priorities[priority].append(item)
75 
76     def _get(self):
77         for bucket in self.priorities:
78             if not bucket:
79                 continue
80             return bucket.popleft()
81 
82 
83 def find_bucket_key(s3_path):
84     """
85     This is a helper function that given an s3 path such that the path is of
86     the form: bucket/key
87     It will return the bucket and the key represented by the s3 path
88     """
89     s3_components = s3_path.split('/')
90     bucket = s3_components[0]
91     s3_key = ""
92     if len(s3_components) > 1:
93         s3_key = '/'.join(s3_components[1:])
94     return bucket, s3_key
95 
96 
97 def split_s3_bucket_key(s3_path):
98     """Split s3 path into bucket and key prefix.
99 
100     This will also handle the s3:// prefix.
101 
102     :return: Tuple of ('bucketname', 'keyname')
103 
104     """
105     if s3_path.startswith('s3://'):
106         s3_path = s3_path[5:]
107     return find_bucket_key(s3_path)
108 
109 
110 def get_file_stat(path):
111     """
112     This is a helper function that given a local path return the size of
113     the file in bytes and time of last modification.
114     """
115     stats = os.stat(path)
116     update_time = datetime.fromtimestamp(stats.st_mtime, tzlocal())
117     return stats.st_size, update_time
118 
119 
120 def check_etag(etag, fileobj):
121     """
122     This fucntion checks the etag and the md5 checksum to ensure no
123     data was corrupted upon transfer.
124     """
125     get_chunk = partial(fileobj.read, 1024 * 1024)
126     m = hashlib.md5()
127     for chunk in iter(get_chunk, b''):
128         m.update(chunk)
129     if '-' not in etag:
130         if etag != m.hexdigest():
131             raise MD5Error
132 
133 
134 def check_error(response_data):
135     """
136     A helper function that prints out the error message recieved in the
137     response_data and raises an error when there is an error.
138     """
139     if response_data:
140         if 'Errors' in response_data:
141             errors = response_data['Errors']
142             for error in errors:
143                 raise Exception("Error: %s\n" % error['Message'])
144 
145 
146 def operate(service, cmd, kwargs):
147     """
148     A helper function that universally calls any command by taking in the
149     service, name of the command, and any additional parameters required in
150     the call.
151     """
152     operation = service.get_operation(cmd)
153     http_response, response_data = operation.call(**kwargs)
154     check_error(response_data)
155     return response_data, http_response
156 
157 
158 def find_chunksize(size, current_chunksize):
159     """
160     The purpose of this function is determine a chunksize so that
161     the number of parts in a multipart upload is not greater than
162     the ``MAX_PARTS``.  If the ``chunksize`` is greater than
163     ``MAX_SINGLE_UPLOAD_SIZE`` it returns ``MAX_SINGLE_UPLOAD_SIZE``.
164     """
165     chunksize = current_chunksize
166     num_parts = int(math.ceil(size / float(chunksize)))
167     while num_parts > MAX_PARTS:
168         chunksize *= 2
169         num_parts = int(math.ceil(size / float(chunksize)))
170     if chunksize > MAX_SINGLE_UPLOAD_SIZE:
171         return MAX_SINGLE_UPLOAD_SIZE
172     else:
173         return chunksize
174 
175 
176 class MultiCounter(object):
177     """
178     This class is used as a way to keep track of how many multipart
179     operations are in progress.  It also is used to track how many
180     part operations are occuring.
181     """
182     def __init__(self):
183         self.count = 0
184 
185 
186 def uni_print(statement):
187     """
188     This function is used to properly write unicode to stdout.  It
189     ensures that the proper encoding is used if the statement is
190     not in a version type of string.  The initial check is to
191     allow if ``sys.stdout`` does not use an encoding
192     """
193     encoding = getattr(sys.stdout, 'encoding', None)
194     if encoding is not None and not PY3:
195         sys.stdout.write(statement.encode(sys.stdout.encoding))
196     else:
197         try:
198             sys.stdout.write(statement)
199         except UnicodeEncodeError:
200             # Some file like objects like cStringIO will
201             # try to decode as ascii.  Interestingly enough
202             # this works with a normal StringIO.
203             sys.stdout.write(statement.encode('utf-8'))
204 
205 
206 def guess_content_type(filename):
207     """Given a filename, guess it's content type.
208 
209     If the type cannot be guessed, a value of None is returned.
210     """
211     return mimetypes.guess_type(filename)[0]
212 
213 
214 def relative_path(filename, start=os.path.curdir):
215     """Cross platform relative path of a filename.
216 
217     If no relative path can be calculated (i.e different
218     drives on Windows), then instead of raising a ValueError,
219     the absolute path is returned.
220 
221     """
222     try:
223         dirname, basename = os.path.split(filename)
224         relative_dir = os.path.relpath(dirname, start)
225         return os.path.join(relative_dir, basename)
226     except ValueError:
227         return os.path.abspath(filename)
228 
229 
230 class ReadFileChunk(object):
231     def __init__(self, filename, start_byte, size):
232         self._filename = filename
233         self._start_byte = start_byte
234         self._fileobj = open(self._filename, 'rb')
235         self._size = self._calculate_file_size(self._fileobj, requested_size=size,
236                                                start_byte=start_byte)
237         self._fileobj.seek(self._start_byte)
238         self._amount_read = 0
239 
240     def _calculate_file_size(self, fileobj, requested_size, start_byte):
241         actual_file_size = os.fstat(fileobj.fileno()).st_size
242         max_chunk_size = actual_file_size - start_byte
243         return min(max_chunk_size, requested_size)
244 
245     def read(self, amount=None):
246         if amount is None:
247             remaining = self._size - self._amount_read
248             data = self._fileobj.read(remaining)
249             self._amount_read += remaining
250             return data
251         else:
252             actual_amount = min(self._size - self._amount_read, amount)
253             data = self._fileobj.read(actual_amount)
254             self._amount_read += actual_amount
255             return data
256 
257     def seek(self, where):
258         self._fileobj.seek(self._start_byte + where)
259         self._amount_read = where
260 
261     def close(self):
262         self._fileobj.close()
263 
264     def tell(self):
265         return self._amount_read
266 
267     def __len__(self):
268         # __len__ is defined because requests will try to determine the length
269         # of the stream to set a content length.  In the normal case
270         # of the file it will just stat the file, but we need to change that
271         # behavior.  By providing a __len__, requests will use that instead
272         # of stat'ing the file.
273         return self._size
274 
275     def __enter__(self):
276         return self
277 
278     def __exit__(self, *args, **kwargs):
279         self._fileobj.close()
280 
281     def __iter__(self):
282         # This is a workaround for http://bugs.python.org/issue17575
283         # Basically httplib will try to iterate over the contents, even
284         # if its a file like object.  This wasn't noticed because we've
285         # already exhausted the stream so iterating over the file immediately
286         # steps, which is what we're simulating here.
287         return iter([])
288 
289 
290 def _date_parser(date_string):
291     return parse(date_string).astimezone(tzlocal())
292 
293 
294 class BucketLister(object):
295     """List keys in a bucket."""
296     def __init__(self, operation, endpoint, date_parser=_date_parser):
297         self._operation = operation
298         self._endpoint = endpoint
299         self._date_parser = date_parser
300 
301     def list_objects(self, bucket, prefix=None):
302         kwargs = {'bucket': bucket, 'encoding_type': 'url'}
303         if prefix is not None:
304             kwargs['prefix'] = prefix
305         pages = self._operation.paginate(self._endpoint, **kwargs)
306         for response, page in pages:
307             contents = page['Contents']
308             for content in contents:
309                 source_path = bucket + '/' + unquote_str(content['Key'])
310                 size = content['Size']
311                 last_update = self._date_parser(content['LastModified'])
312                 yield source_path, size, last_update
313 
314 
315 IORequest = namedtuple('IORequest', ['filename', 'offset', 'data'])
316 # Used to signal that IO for the filename is finished, and that
317 # any associated resources may be cleaned up.
318 IOCloseRequest = namedtuple('IOCloseRequest', ['filename'])
319
```
--- 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/awscli/customizations/s3/utils.py b/awscli/customizations/s3/utils.py
--- a/awscli/customizations/s3/utils.py
+++ b/awscli/customizations/s3/utils.py
@@ -302,14 +302,39 @@
         kwargs = {'bucket': bucket, 'encoding_type': 'url'}
         if prefix is not None:
             kwargs['prefix'] = prefix
-        pages = self._operation.paginate(self._endpoint, **kwargs)
-        for response, page in pages:
-            contents = page['Contents']
-            for content in contents:
-                source_path = bucket + '/' + unquote_str(content['Key'])
-                size = content['Size']
-                last_update = self._date_parser(content['LastModified'])
-                yield source_path, size, last_update
+        # This event handler is needed because we use encoding_type url and
+        # we're paginating.  The pagination token is the last Key of the
+        # Contents list.  However, botocore does not know that the encoding
+        # type needs to be urldecoded.
+        with ScopedEventHandler(self._operation.session, 'after-call.s3.ListObjects',
+                                self._decode_keys):
+            pages = self._operation.paginate(self._endpoint, **kwargs)
+            for response, page in pages:
+                contents = page['Contents']
+                for content in contents:
+                    source_path = bucket + '/' + content['Key']
+                    size = content['Size']
+                    last_update = self._date_parser(content['LastModified'])
+                    yield source_path, size, last_update
+
+    def _decode_keys(self, parsed, **kwargs):
+        for content in parsed['Contents']:
+            content['Key'] = unquote_str(content['Key'])
+
+
+class ScopedEventHandler(object):
+    """Register an event callback for the duration of a scope."""
+
+    def __init__(self, session, event_name, handler):
+        self._session = session
+        self._event_name = event_name
+        self._handler = handler
+
+    def __enter__(self):
+        self._session.register(self._event_name, self._handler)
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self._session.unregister(self._event_name, self._handler)
 
 
 IORequest = namedtuple('IORequest', ['filename', 'offset', 'data'])

{"golden_diff": "diff --git a/awscli/customizations/s3/utils.py b/awscli/customizations/s3/utils.py\n--- a/awscli/customizations/s3/utils.py\n+++ b/awscli/customizations/s3/utils.py\n@@ -302,14 +302,39 @@\n         kwargs = {'bucket': bucket, 'encoding_type': 'url'}\n         if prefix is not None:\n             kwargs['prefix'] = prefix\n-        pages = self._operation.paginate(self._endpoint, **kwargs)\n-        for response, page in pages:\n-            contents = page['Contents']\n-            for content in contents:\n-                source_path = bucket + '/' + unquote_str(content['Key'])\n-                size = content['Size']\n-                last_update = self._date_parser(content['LastModified'])\n-                yield source_path, size, last_update\n+        # This event handler is needed because we use encoding_type url and\n+        # we're paginating.  The pagination token is the last Key of the\n+        # Contents list.  However, botocore does not know that the encoding\n+        # type needs to be urldecoded.\n+        with ScopedEventHandler(self._operation.session, 'after-call.s3.ListObjects',\n+                                self._decode_keys):\n+            pages = self._operation.paginate(self._endpoint, **kwargs)\n+            for response, page in pages:\n+                contents = page['Contents']\n+                for content in contents:\n+                    source_path = bucket + '/' + content['Key']\n+                    size = content['Size']\n+                    last_update = self._date_parser(content['LastModified'])\n+                    yield source_path, size, last_update\n+\n+    def _decode_keys(self, parsed, **kwargs):\n+        for content in parsed['Contents']:\n+            content['Key'] = unquote_str(content['Key'])\n+\n+\n+class ScopedEventHandler(object):\n+    \"\"\"Register an event callback for the duration of a scope.\"\"\"\n+\n+    def __init__(self, session, event_name, handler):\n+        self._session = session\n+        self._event_name = event_name\n+        self._handler = handler\n+\n+    def __enter__(self):\n+        self._session.register(self._event_name, self._handler)\n+\n+    def __exit__(self, exc_type, exc_value, traceback):\n+        self._session.unregister(self._event_name, self._handler)\n \n \n IORequest = namedtuple('IORequest', ['filename', 'offset', 'data'])\n", "issue": "s3 sync from local to s3 bucket re-uploads files with version 1.3.6\nIt is probably related to #718 but it is still happening on my OSX Mavericks with Python 2.7. I sync local files to S3 using \"aws s3 sync\" but some files always want to re-upload. I did a debug and comparator thinks that \"file does not exist at destination\".\n\nI even removed all scripts and python packages for awscli, botocore and jmespath, and then reinstalled latest 1.3.6 but no luck. I use awscli for s3 sync since version 1.0 without any issues and have only started to experiences such issues starting with 1.3.2.\n\nAppreciate if you could look into it. Let me know if additional information was needed.\n\n", "before_files": [{"content": "# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"). You\n# may not use this file except in compliance with the License. A copy of\n# the License is located at\n#\n#     http://aws.amazon.com/apache2.0/\n#\n# or in the \"license\" file accompanying this file. This file is\n# distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific\n# language governing permissions and limitations under the License.\nfrom datetime import datetime\nimport mimetypes\nimport hashlib\nimport math\nimport os\nimport sys\nfrom collections import namedtuple, deque\nfrom functools import partial\n\nfrom six import PY3\nfrom six.moves import queue\nfrom dateutil.parser import parse\nfrom dateutil.tz import tzlocal\nfrom botocore.compat import unquote_str\n\nfrom awscli.customizations.s3.constants import MAX_PARTS\nfrom awscli.customizations.s3.constants import MAX_SINGLE_UPLOAD_SIZE\n\n\nclass MD5Error(Exception):\n    \"\"\"\n    Exception for md5's that do not match.\n    \"\"\"\n    pass\n\n\nclass StablePriorityQueue(queue.Queue):\n    \"\"\"Priority queue that maintains FIFO order for same priority items.\n\n    This class was written to handle the tasks created in\n    awscli.customizations.s3.tasks, but it's possible to use this\n    class outside of that context.  In order for this to be the case,\n    the following conditions should be met:\n\n        * Objects that are queued should have a PRIORITY attribute.\n          This should be an integer value not to exceed the max_priority\n          value passed into the ``__init__``.  Objects with lower\n          priority numbers are retrieved before objects with higher\n          priority numbers.\n        * A relatively small max_priority should be chosen.  ``get()``\n          calls are O(max_priority).\n\n    Any object that does not have a ``PRIORITY`` attribute or whose\n    priority exceeds ``max_priority`` will be queued at the highest\n    (least important) priority available.\n\n    \"\"\"\n    def __init__(self, maxsize=0, max_priority=20):\n        queue.Queue.__init__(self, maxsize=maxsize)\n        self.priorities = [deque([]) for i in range(max_priority + 1)]\n        self.default_priority = max_priority\n\n    def _qsize(self):\n        size = 0\n        for bucket in self.priorities:\n            size += len(bucket)\n        return size\n\n    def _put(self, item):\n        priority = min(getattr(item, 'PRIORITY', self.default_priority),\n                        self.default_priority)\n        self.priorities[priority].append(item)\n\n    def _get(self):\n        for bucket in self.priorities:\n            if not bucket:\n                continue\n            return bucket.popleft()\n\n\ndef find_bucket_key(s3_path):\n    \"\"\"\n    This is a helper function that given an s3 path such that the path is of\n    the form: bucket/key\n    It will return the bucket and the key represented by the s3 path\n    \"\"\"\n    s3_components = s3_path.split('/')\n    bucket = s3_components[0]\n    s3_key = \"\"\n    if len(s3_components) > 1:\n        s3_key = '/'.join(s3_components[1:])\n    return bucket, s3_key\n\n\ndef split_s3_bucket_key(s3_path):\n    \"\"\"Split s3 path into bucket and key prefix.\n\n    This will also handle the s3:// prefix.\n\n    :return: Tuple of ('bucketname', 'keyname')\n\n    \"\"\"\n    if s3_path.startswith('s3://'):\n        s3_path = s3_path[5:]\n    return find_bucket_key(s3_path)\n\n\ndef get_file_stat(path):\n    \"\"\"\n    This is a helper function that given a local path return the size of\n    the file in bytes and time of last modification.\n    \"\"\"\n    stats = os.stat(path)\n    update_time = datetime.fromtimestamp(stats.st_mtime, tzlocal())\n    return stats.st_size, update_time\n\n\ndef check_etag(etag, fileobj):\n    \"\"\"\n    This fucntion checks the etag and the md5 checksum to ensure no\n    data was corrupted upon transfer.\n    \"\"\"\n    get_chunk = partial(fileobj.read, 1024 * 1024)\n    m = hashlib.md5()\n    for chunk in iter(get_chunk, b''):\n        m.update(chunk)\n    if '-' not in etag:\n        if etag != m.hexdigest():\n            raise MD5Error\n\n\ndef check_error(response_data):\n    \"\"\"\n    A helper function that prints out the error message recieved in the\n    response_data and raises an error when there is an error.\n    \"\"\"\n    if response_data:\n        if 'Errors' in response_data:\n            errors = response_data['Errors']\n            for error in errors:\n                raise Exception(\"Error: %s\\n\" % error['Message'])\n\n\ndef operate(service, cmd, kwargs):\n    \"\"\"\n    A helper function that universally calls any command by taking in the\n    service, name of the command, and any additional parameters required in\n    the call.\n    \"\"\"\n    operation = service.get_operation(cmd)\n    http_response, response_data = operation.call(**kwargs)\n    check_error(response_data)\n    return response_data, http_response\n\n\ndef find_chunksize(size, current_chunksize):\n    \"\"\"\n    The purpose of this function is determine a chunksize so that\n    the number of parts in a multipart upload is not greater than\n    the ``MAX_PARTS``.  If the ``chunksize`` is greater than\n    ``MAX_SINGLE_UPLOAD_SIZE`` it returns ``MAX_SINGLE_UPLOAD_SIZE``.\n    \"\"\"\n    chunksize = current_chunksize\n    num_parts = int(math.ceil(size / float(chunksize)))\n    while num_parts > MAX_PARTS:\n        chunksize *= 2\n        num_parts = int(math.ceil(size / float(chunksize)))\n    if chunksize > MAX_SINGLE_UPLOAD_SIZE:\n        return MAX_SINGLE_UPLOAD_SIZE\n    else:\n        return chunksize\n\n\nclass MultiCounter(object):\n    \"\"\"\n    This class is used as a way to keep track of how many multipart\n    operations are in progress.  It also is used to track how many\n    part operations are occuring.\n    \"\"\"\n    def __init__(self):\n        self.count = 0\n\n\ndef uni_print(statement):\n    \"\"\"\n    This function is used to properly write unicode to stdout.  It\n    ensures that the proper encoding is used if the statement is\n    not in a version type of string.  The initial check is to\n    allow if ``sys.stdout`` does not use an encoding\n    \"\"\"\n    encoding = getattr(sys.stdout, 'encoding', None)\n    if encoding is not None and not PY3:\n        sys.stdout.write(statement.encode(sys.stdout.encoding))\n    else:\n        try:\n            sys.stdout.write(statement)\n        except UnicodeEncodeError:\n            # Some file like objects like cStringIO will\n            # try to decode as ascii.  Interestingly enough\n            # this works with a normal StringIO.\n            sys.stdout.write(statement.encode('utf-8'))\n\n\ndef guess_content_type(filename):\n    \"\"\"Given a filename, guess it's content type.\n\n    If the type cannot be guessed, a value of None is returned.\n    \"\"\"\n    return mimetypes.guess_type(filename)[0]\n\n\ndef relative_path(filename, start=os.path.curdir):\n    \"\"\"Cross platform relative path of a filename.\n\n    If no relative path can be calculated (i.e different\n    drives on Windows), then instead of raising a ValueError,\n    the absolute path is returned.\n\n    \"\"\"\n    try:\n        dirname, basename = os.path.split(filename)\n        relative_dir = os.path.relpath(dirname, start)\n        return os.path.join(relative_dir, basename)\n    except ValueError:\n        return os.path.abspath(filename)\n\n\nclass ReadFileChunk(object):\n    def __init__(self, filename, start_byte, size):\n        self._filename = filename\n        self._start_byte = start_byte\n        self._fileobj = open(self._filename, 'rb')\n        self._size = self._calculate_file_size(self._fileobj, requested_size=size,\n                                               start_byte=start_byte)\n        self._fileobj.seek(self._start_byte)\n        self._amount_read = 0\n\n    def _calculate_file_size(self, fileobj, requested_size, start_byte):\n        actual_file_size = os.fstat(fileobj.fileno()).st_size\n        max_chunk_size = actual_file_size - start_byte\n        return min(max_chunk_size, requested_size)\n\n    def read(self, amount=None):\n        if amount is None:\n            remaining = self._size - self._amount_read\n            data = self._fileobj.read(remaining)\n            self._amount_read += remaining\n            return data\n        else:\n            actual_amount = min(self._size - self._amount_read, amount)\n            data = self._fileobj.read(actual_amount)\n            self._amount_read += actual_amount\n            return data\n\n    def seek(self, where):\n        self._fileobj.seek(self._start_byte + where)\n        self._amount_read = where\n\n    def close(self):\n        self._fileobj.close()\n\n    def tell(self):\n        return self._amount_read\n\n    def __len__(self):\n        # __len__ is defined because requests will try to determine the length\n        # of the stream to set a content length.  In the normal case\n        # of the file it will just stat the file, but we need to change that\n        # behavior.  By providing a __len__, requests will use that instead\n        # of stat'ing the file.\n        return self._size\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *args, **kwargs):\n        self._fileobj.close()\n\n    def __iter__(self):\n        # This is a workaround for http://bugs.python.org/issue17575\n        # Basically httplib will try to iterate over the contents, even\n        # if its a file like object.  This wasn't noticed because we've\n        # already exhausted the stream so iterating over the file immediately\n        # steps, which is what we're simulating here.\n        return iter([])\n\n\ndef _date_parser(date_string):\n    return parse(date_string).astimezone(tzlocal())\n\n\nclass BucketLister(object):\n    \"\"\"List keys in a bucket.\"\"\"\n    def __init__(self, operation, endpoint, date_parser=_date_parser):\n        self._operation = operation\n        self._endpoint = endpoint\n        self._date_parser = date_parser\n\n    def list_objects(self, bucket, prefix=None):\n        kwargs = {'bucket': bucket, 'encoding_type': 'url'}\n        if prefix is not None:\n            kwargs['prefix'] = prefix\n        pages = self._operation.paginate(self._endpoint, **kwargs)\n        for response, page in pages:\n            contents = page['Contents']\n            for content in contents:\n                source_path = bucket + '/' + unquote_str(content['Key'])\n                size = content['Size']\n                last_update = self._date_parser(content['LastModified'])\n                yield source_path, size, last_update\n\n\nIORequest = namedtuple('IORequest', ['filename', 'offset', 'data'])\n# Used to signal that IO for the filename is finished, and that\n# any associated resources may be cleaned up.\nIOCloseRequest = namedtuple('IOCloseRequest', ['filename'])\n", "path": "awscli/customizations/s3/utils.py"}], "after_files": [{"content": "# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"). You\n# may not use this file except in compliance with the License. A copy of\n# the License is located at\n#\n#     http://aws.amazon.com/apache2.0/\n#\n# or in the \"license\" file accompanying this file. This file is\n# distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific\n# language governing permissions and limitations under the License.\nfrom datetime import datetime\nimport mimetypes\nimport hashlib\nimport math\nimport os\nimport sys\nfrom collections import namedtuple, deque\nfrom functools import partial\n\nfrom six import PY3\nfrom six.moves import queue\nfrom dateutil.parser import parse\nfrom dateutil.tz import tzlocal\nfrom botocore.compat import unquote_str\n\nfrom awscli.customizations.s3.constants import MAX_PARTS\nfrom awscli.customizations.s3.constants import MAX_SINGLE_UPLOAD_SIZE\n\n\nclass MD5Error(Exception):\n    \"\"\"\n    Exception for md5's that do not match.\n    \"\"\"\n    pass\n\n\nclass StablePriorityQueue(queue.Queue):\n    \"\"\"Priority queue that maintains FIFO order for same priority items.\n\n    This class was written to handle the tasks created in\n    awscli.customizations.s3.tasks, but it's possible to use this\n    class outside of that context.  In order for this to be the case,\n    the following conditions should be met:\n\n        * Objects that are queued should have a PRIORITY attribute.\n          This should be an integer value not to exceed the max_priority\n          value passed into the ``__init__``.  Objects with lower\n          priority numbers are retrieved before objects with higher\n          priority numbers.\n        * A relatively small max_priority should be chosen.  ``get()``\n          calls are O(max_priority).\n\n    Any object that does not have a ``PRIORITY`` attribute or whose\n    priority exceeds ``max_priority`` will be queued at the highest\n    (least important) priority available.\n\n    \"\"\"\n    def __init__(self, maxsize=0, max_priority=20):\n        queue.Queue.__init__(self, maxsize=maxsize)\n        self.priorities = [deque([]) for i in range(max_priority + 1)]\n        self.default_priority = max_priority\n\n    def _qsize(self):\n        size = 0\n        for bucket in self.priorities:\n            size += len(bucket)\n        return size\n\n    def _put(self, item):\n        priority = min(getattr(item, 'PRIORITY', self.default_priority),\n                        self.default_priority)\n        self.priorities[priority].append(item)\n\n    def _get(self):\n        for bucket in self.priorities:\n            if not bucket:\n                continue\n            return bucket.popleft()\n\n\ndef find_bucket_key(s3_path):\n    \"\"\"\n    This is a helper function that given an s3 path such that the path is of\n    the form: bucket/key\n    It will return the bucket and the key represented by the s3 path\n    \"\"\"\n    s3_components = s3_path.split('/')\n    bucket = s3_components[0]\n    s3_key = \"\"\n    if len(s3_components) > 1:\n        s3_key = '/'.join(s3_components[1:])\n    return bucket, s3_key\n\n\ndef split_s3_bucket_key(s3_path):\n    \"\"\"Split s3 path into bucket and key prefix.\n\n    This will also handle the s3:// prefix.\n\n    :return: Tuple of ('bucketname', 'keyname')\n\n    \"\"\"\n    if s3_path.startswith('s3://'):\n        s3_path = s3_path[5:]\n    return find_bucket_key(s3_path)\n\n\ndef get_file_stat(path):\n    \"\"\"\n    This is a helper function that given a local path return the size of\n    the file in bytes and time of last modification.\n    \"\"\"\n    stats = os.stat(path)\n    update_time = datetime.fromtimestamp(stats.st_mtime, tzlocal())\n    return stats.st_size, update_time\n\n\ndef check_etag(etag, fileobj):\n    \"\"\"\n    This fucntion checks the etag and the md5 checksum to ensure no\n    data was corrupted upon transfer.\n    \"\"\"\n    get_chunk = partial(fileobj.read, 1024 * 1024)\n    m = hashlib.md5()\n    for chunk in iter(get_chunk, b''):\n        m.update(chunk)\n    if '-' not in etag:\n        if etag != m.hexdigest():\n            raise MD5Error\n\n\ndef check_error(response_data):\n    \"\"\"\n    A helper function that prints out the error message recieved in the\n    response_data and raises an error when there is an error.\n    \"\"\"\n    if response_data:\n        if 'Errors' in response_data:\n            errors = response_data['Errors']\n            for error in errors:\n                raise Exception(\"Error: %s\\n\" % error['Message'])\n\n\ndef operate(service, cmd, kwargs):\n    \"\"\"\n    A helper function that universally calls any command by taking in the\n    service, name of the command, and any additional parameters required in\n    the call.\n    \"\"\"\n    operation = service.get_operation(cmd)\n    http_response, response_data = operation.call(**kwargs)\n    check_error(response_data)\n    return response_data, http_response\n\n\ndef find_chunksize(size, current_chunksize):\n    \"\"\"\n    The purpose of this function is determine a chunksize so that\n    the number of parts in a multipart upload is not greater than\n    the ``MAX_PARTS``.  If the ``chunksize`` is greater than\n    ``MAX_SINGLE_UPLOAD_SIZE`` it returns ``MAX_SINGLE_UPLOAD_SIZE``.\n    \"\"\"\n    chunksize = current_chunksize\n    num_parts = int(math.ceil(size / float(chunksize)))\n    while num_parts > MAX_PARTS:\n        chunksize *= 2\n        num_parts = int(math.ceil(size / float(chunksize)))\n    if chunksize > MAX_SINGLE_UPLOAD_SIZE:\n        return MAX_SINGLE_UPLOAD_SIZE\n    else:\n        return chunksize\n\n\nclass MultiCounter(object):\n    \"\"\"\n    This class is used as a way to keep track of how many multipart\n    operations are in progress.  It also is used to track how many\n    part operations are occuring.\n    \"\"\"\n    def __init__(self):\n        self.count = 0\n\n\ndef uni_print(statement):\n    \"\"\"\n    This function is used to properly write unicode to stdout.  It\n    ensures that the proper encoding is used if the statement is\n    not in a version type of string.  The initial check is to\n    allow if ``sys.stdout`` does not use an encoding\n    \"\"\"\n    encoding = getattr(sys.stdout, 'encoding', None)\n    if encoding is not None and not PY3:\n        sys.stdout.write(statement.encode(sys.stdout.encoding))\n    else:\n        try:\n            sys.stdout.write(statement)\n        except UnicodeEncodeError:\n            # Some file like objects like cStringIO will\n            # try to decode as ascii.  Interestingly enough\n            # this works with a normal StringIO.\n            sys.stdout.write(statement.encode('utf-8'))\n\n\ndef guess_content_type(filename):\n    \"\"\"Given a filename, guess it's content type.\n\n    If the type cannot be guessed, a value of None is returned.\n    \"\"\"\n    return mimetypes.guess_type(filename)[0]\n\n\ndef relative_path(filename, start=os.path.curdir):\n    \"\"\"Cross platform relative path of a filename.\n\n    If no relative path can be calculated (i.e different\n    drives on Windows), then instead of raising a ValueError,\n    the absolute path is returned.\n\n    \"\"\"\n    try:\n        dirname, basename = os.path.split(filename)\n        relative_dir = os.path.relpath(dirname, start)\n        return os.path.join(relative_dir, basename)\n    except ValueError:\n        return os.path.abspath(filename)\n\n\nclass ReadFileChunk(object):\n    def __init__(self, filename, start_byte, size):\n        self._filename = filename\n        self._start_byte = start_byte\n        self._fileobj = open(self._filename, 'rb')\n        self._size = self._calculate_file_size(self._fileobj, requested_size=size,\n                                               start_byte=start_byte)\n        self._fileobj.seek(self._start_byte)\n        self._amount_read = 0\n\n    def _calculate_file_size(self, fileobj, requested_size, start_byte):\n        actual_file_size = os.fstat(fileobj.fileno()).st_size\n        max_chunk_size = actual_file_size - start_byte\n        return min(max_chunk_size, requested_size)\n\n    def read(self, amount=None):\n        if amount is None:\n            remaining = self._size - self._amount_read\n            data = self._fileobj.read(remaining)\n            self._amount_read += remaining\n            return data\n        else:\n            actual_amount = min(self._size - self._amount_read, amount)\n            data = self._fileobj.read(actual_amount)\n            self._amount_read += actual_amount\n            return data\n\n    def seek(self, where):\n        self._fileobj.seek(self._start_byte + where)\n        self._amount_read = where\n\n    def close(self):\n        self._fileobj.close()\n\n    def tell(self):\n        return self._amount_read\n\n    def __len__(self):\n        # __len__ is defined because requests will try to determine the length\n        # of the stream to set a content length.  In the normal case\n        # of the file it will just stat the file, but we need to change that\n        # behavior.  By providing a __len__, requests will use that instead\n        # of stat'ing the file.\n        return self._size\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *args, **kwargs):\n        self._fileobj.close()\n\n    def __iter__(self):\n        # This is a workaround for http://bugs.python.org/issue17575\n        # Basically httplib will try to iterate over the contents, even\n        # if its a file like object.  This wasn't noticed because we've\n        # already exhausted the stream so iterating over the file immediately\n        # steps, which is what we're simulating here.\n        return iter([])\n\n\ndef _date_parser(date_string):\n    return parse(date_string).astimezone(tzlocal())\n\n\nclass BucketLister(object):\n    \"\"\"List keys in a bucket.\"\"\"\n    def __init__(self, operation, endpoint, date_parser=_date_parser):\n        self._operation = operation\n        self._endpoint = endpoint\n        self._date_parser = date_parser\n\n    def list_objects(self, bucket, prefix=None):\n        kwargs = {'bucket': bucket, 'encoding_type': 'url'}\n        if prefix is not None:\n            kwargs['prefix'] = prefix\n        # This event handler is needed because we use encoding_type url and\n        # we're paginating.  The pagination token is the last Key of the\n        # Contents list.  However, botocore does not know that the encoding\n        # type needs to be urldecoded.\n        with ScopedEventHandler(self._operation.session, 'after-call.s3.ListObjects',\n                                self._decode_keys):\n            pages = self._operation.paginate(self._endpoint, **kwargs)\n            for response, page in pages:\n                contents = page['Contents']\n                for content in contents:\n                    source_path = bucket + '/' + content['Key']\n                    size = content['Size']\n                    last_update = self._date_parser(content['LastModified'])\n                    yield source_path, size, last_update\n\n    def _decode_keys(self, parsed, **kwargs):\n        for content in parsed['Contents']:\n            content['Key'] = unquote_str(content['Key'])\n\n\nclass ScopedEventHandler(object):\n    \"\"\"Register an event callback for the duration of a scope.\"\"\"\n\n    def __init__(self, session, event_name, handler):\n        self._session = session\n        self._event_name = event_name\n        self._handler = handler\n\n    def __enter__(self):\n        self._session.register(self._event_name, self._handler)\n\n    def __exit__(self, exc_type, exc_value, traceback):\n        self._session.unregister(self._event_name, self._handler)\n\n\nIORequest = namedtuple('IORequest', ['filename', 'offset', 'data'])\n# Used to signal that IO for the filename is finished, and that\n# any associated resources may be cleaned up.\nIOCloseRequest = namedtuple('IOCloseRequest', ['filename'])\n", "path": "awscli/customizations/s3/utils.py"}]}