Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/stripe/multipart_data_generator.py: 19%
58 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1from __future__ import absolute_import, division, print_function
3import random
4import io
6import stripe
9class MultipartDataGenerator(object):
10 def __init__(self, chunk_size=1028):
11 self.data = io.BytesIO()
12 self.line_break = "\r\n"
13 self.boundary = self._initialize_boundary()
14 self.chunk_size = chunk_size
16 def add_params(self, params):
17 # Flatten parameters first
18 params = dict(stripe.api_requestor._api_encode(params))
20 for key, value in stripe.six.iteritems(params):
21 if value is None:
22 continue
24 self._write(self.param_header())
25 self._write(self.line_break)
26 if hasattr(value, "read"):
27 filename = "blob"
28 if hasattr(value, "name"):
29 # Convert the filename to string, just in case it's not
30 # already one. E.g. `tempfile.TemporaryFile` has a `name`
31 # attribute but it's an `int`.
32 filename = stripe.six.text_type(value.name)
34 self._write('Content-Disposition: form-data; name="')
35 self._write(key)
36 self._write('"; filename="')
37 self._write(filename)
38 self._write('"')
39 self._write(self.line_break)
40 self._write("Content-Type: application/octet-stream")
41 self._write(self.line_break)
42 self._write(self.line_break)
44 self._write_file(value)
45 else:
46 self._write('Content-Disposition: form-data; name="')
47 self._write(key)
48 self._write('"')
49 self._write(self.line_break)
50 self._write(self.line_break)
51 self._write(str(value))
53 self._write(self.line_break)
55 def param_header(self):
56 return "--%s" % self.boundary
58 def get_post_data(self):
59 self._write("--%s--" % (self.boundary,))
60 self._write(self.line_break)
61 return self.data.getvalue()
63 def _write(self, value):
64 if isinstance(value, stripe.six.binary_type):
65 array = bytearray(value)
66 elif isinstance(value, stripe.six.text_type):
67 array = bytearray(value, encoding="utf-8")
68 else:
69 raise TypeError(
70 "unexpected type: {value_type}".format(value_type=type(value))
71 )
73 self.data.write(array)
75 def _write_file(self, f):
76 while True:
77 file_contents = f.read(self.chunk_size)
78 if not file_contents:
79 break
80 self._write(file_contents)
82 def _initialize_boundary(self):
83 return random.randint(0, 2 ** 63)