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
|
import random
import io
from stripe._encode import _api_encode
class MultipartDataGenerator(object):
data: io.BytesIO
line_break: str
boundary: int
chunk_size: int
def __init__(self, chunk_size: int = 1028):
self.data = io.BytesIO()
self.line_break = "\r\n"
self.boundary = self._initialize_boundary()
self.chunk_size = chunk_size
def add_params(self, params):
# Flatten parameters first
params = dict(_api_encode(params, "V1"))
for key, value in params.items():
if value is None:
continue
self._write(self.param_header())
self._write(self.line_break)
if hasattr(value, "read"):
filename = "blob"
if hasattr(value, "name"):
# Convert the filename to string, just in case it's not
# already one. E.g. `tempfile.TemporaryFile` has a `name`
# attribute but it's an `int`.
filename = str(value.name)
self._write('Content-Disposition: form-data; name="')
self._write(key)
self._write('"; filename="')
self._write(filename)
self._write('"')
self._write(self.line_break)
self._write("Content-Type: application/octet-stream")
self._write(self.line_break)
self._write(self.line_break)
self._write_file(value)
else:
self._write('Content-Disposition: form-data; name="')
self._write(key)
self._write('"')
self._write(self.line_break)
self._write(self.line_break)
self._write(str(value))
self._write(self.line_break)
def param_header(self):
return "--%s" % self.boundary
def get_post_data(self):
self._write("--%s--" % (self.boundary,))
self._write(self.line_break)
return self.data.getvalue()
def _write(self, value):
if isinstance(value, bytes):
array = bytearray(value)
elif isinstance(value, str):
array = bytearray(value, encoding="utf-8")
else:
raise TypeError(
"unexpected type: {value_type}".format(value_type=type(value))
)
self.data.write(array)
def _write_file(self, f):
while True:
file_contents = f.read(self.chunk_size)
if not file_contents:
break
self._write(file_contents)
def _initialize_boundary(self):
return random.randint(0, 2**63)
|
