-
Notifications
You must be signed in to change notification settings - Fork 925
/
files.py
115 lines (90 loc) · 3.27 KB
/
files.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import mimetypes
from libcloud.utils.py3 import PY3
from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import next
from libcloud.utils.py3 import b
CHUNK_SIZE = 8096
if PY3:
from io import FileIO as file
def read_in_chunks(iterator, chunk_size=None, fill_size=False):
"""
Return a generator which yields data in chunks.
@type iterator: C{Iterator}
@param response: An object which implements an iterator interface
or a File like object with read method.
@type chunk_size: C{int}
@param chunk_size: Optional chunk size (defaults to CHUNK_SIZE)
@type fill_size: C{bool}
@param fill_size: If True, make sure chunks are chunk_size in length
(except for last chunk).
TODO: At some point in the future we could use byte arrays here if version
>= Python 3. This should speed things up a bit and reduce memory usage.
"""
chunk_size = chunk_size or CHUNK_SIZE
if isinstance(iterator, (file, httplib.HTTPResponse)):
get_data = iterator.read
args = (chunk_size, )
else:
get_data = next
args = (iterator, )
data = b('')
empty = False
while not empty or len(data) > 0:
if not empty:
try:
chunk = b(get_data(*args))
if len(chunk) > 0:
data += chunk
else:
empty = True
except StopIteration:
empty = True
if len(data) == 0:
raise StopIteration
if fill_size:
if empty or len(data) >= chunk_size:
yield data[:chunk_size]
data = data[chunk_size:]
else:
yield data
data = b('')
def exhaust_iterator(iterator):
"""
Exhaust an iterator and return all data returned by it.
@type iterator: C{Iterator}
@param response: An object which implements an iterator interface
or a File like object with read method.
@rtype C{str}
@return Data returned by the iterator.
"""
data = b('')
try:
chunk = b(next(iterator))
except StopIteration:
chunk = b('')
while len(chunk) > 0:
data += chunk
try:
chunk = b(next(iterator))
except StopIteration:
chunk = b('')
return data
def guess_file_mime_type(file_path):
filename = os.path.basename(file_path)
(mimetype, encoding) = mimetypes.guess_type(filename)
return mimetype, encoding