Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/stripe/multipart_data_generator.py: 19%

1from __future__ import absolute_import, division, print_function 

2 

3import random 

4import io 

5 

6import stripe 

7 

8 

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 

15 

16 def add_params(self, params): 

17 # Flatten parameters first 

18 params = dict(stripe.api_requestor._api_encode(params)) 

19 

20 for key, value in stripe.six.iteritems(params): 

21 if value is None: 

22 continue 

23 

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) 

33 

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) 

43 

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)) 

52 

53 self._write(self.line_break) 

54 

55 def param_header(self): 

56 return "--%s" % self.boundary 

57 

58 def get_post_data(self): 

59 self._write("--%s--" % (self.boundary,)) 

60 self._write(self.line_break) 

61 return self.data.getvalue() 

62 

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 ) 

72 

73 self.data.write(array) 

74 

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) 

81 

82 def _initialize_boundary(self): 

83 return random.randint(0, 2 ** 63)