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

1from __future__ import absolute_import, division, print_function 

2 

3from stripe import api_requestor, six, util 

4from stripe.stripe_object import StripeObject 

5 

6from stripe.six.moves.urllib.parse import quote_plus 

7 

8 

9class ListObject(StripeObject): 

10 OBJECT_NAME = "list" 

11 

12 def list( 

13 self, api_key=None, stripe_version=None, stripe_account=None, **params 

14 ): 

15 stripe_object = self._request( 

16 "get", 

17 self.get("url"), 

18 api_key=api_key, 

19 stripe_version=stripe_version, 

20 stripe_account=stripe_account, 

21 **params 

22 ) 

23 stripe_object._retrieve_params = params 

24 return stripe_object 

25 

26 def create( 

27 self, 

28 api_key=None, 

29 idempotency_key=None, 

30 stripe_version=None, 

31 stripe_account=None, 

32 **params 

33 ): 

34 return self._request( 

35 "post", 

36 self.get("url"), 

37 api_key=api_key, 

38 idempotency_key=idempotency_key, 

39 stripe_version=stripe_version, 

40 stripe_account=stripe_account, 

41 **params 

42 ) 

43 

44 def retrieve( 

45 self, 

46 id, 

47 api_key=None, 

48 stripe_version=None, 

49 stripe_account=None, 

50 **params 

51 ): 

52 url = "%s/%s" % (self.get("url"), quote_plus(util.utf8(id))) 

53 return self._request( 

54 "get", 

55 url, 

56 api_key=api_key, 

57 stripe_version=stripe_version, 

58 stripe_account=stripe_account, 

59 **params 

60 ) 

61 

62 def _request( 

63 self, 

64 method_, 

65 url_, 

66 api_key=None, 

67 idempotency_key=None, 

68 stripe_version=None, 

69 stripe_account=None, 

70 **params 

71 ): 

72 api_key = api_key or self.api_key 

73 stripe_version = stripe_version or self.stripe_version 

74 stripe_account = stripe_account or self.stripe_account 

75 

76 requestor = api_requestor.APIRequestor( 

77 api_key, api_version=stripe_version, account=stripe_account 

78 ) 

79 headers = util.populate_headers(idempotency_key) 

80 response, api_key = requestor.request(method_, url_, params, headers) 

81 stripe_object = util.convert_to_stripe_object( 

82 response, api_key, stripe_version, stripe_account 

83 ) 

84 return stripe_object 

85 

86 def __getitem__(self, k): 

87 if isinstance(k, six.string_types): 

88 return super(ListObject, self).__getitem__(k) 

89 else: 

90 raise KeyError( 

91 "You tried to access the %s index, but ListObject types only " 

92 "support string keys. (HINT: List calls return an object with " 

93 "a 'data' (which is the data array). You likely want to call " 

94 ".data[%s])" % (repr(k), repr(k)) 

95 ) 

96 

97 def __iter__(self): 

98 return getattr(self, "data", []).__iter__() 

99 

100 def __len__(self): 

101 return getattr(self, "data", []).__len__() 

102 

103 def __reversed__(self): 

104 return getattr(self, "data", []).__reversed__() 

105 

106 def auto_paging_iter(self): 

107 page = self 

108 

109 while True: 

110 if ( 

111 "ending_before" in self._retrieve_params 

112 and "starting_after" not in self._retrieve_params 

113 ): 

114 for item in reversed(page): 

115 yield item 

116 page = page.previous_page() 

117 else: 

118 for item in page: 

119 yield item 

120 page = page.next_page() 

121 

122 if page.is_empty: 

123 break 

124 

125 @classmethod 

126 def empty_list( 

127 cls, api_key=None, stripe_version=None, stripe_account=None 

128 ): 

129 return cls.construct_from( 

130 {"data": []}, 

131 key=api_key, 

132 stripe_version=stripe_version, 

133 stripe_account=stripe_account, 

134 last_response=None, 

135 ) 

136 

137 @property 

138 def is_empty(self): 

139 return not self.data 

140 

141 def next_page( 

142 self, api_key=None, stripe_version=None, stripe_account=None, **params 

143 ): 

144 if not self.has_more: 

145 return self.empty_list( 

146 api_key=api_key, 

147 stripe_version=stripe_version, 

148 stripe_account=stripe_account, 

149 ) 

150 

151 last_id = self.data[-1].id 

152 

153 params_with_filters = self._retrieve_params.copy() 

154 params_with_filters.update({"starting_after": last_id}) 

155 params_with_filters.update(params) 

156 

157 return self.list( 

158 api_key=api_key, 

159 stripe_version=stripe_version, 

160 stripe_account=stripe_account, 

161 **params_with_filters 

162 ) 

163 

164 def previous_page( 

165 self, api_key=None, stripe_version=None, stripe_account=None, **params 

166 ): 

167 if not self.has_more: 

168 return self.empty_list( 

169 api_key=api_key, 

170 stripe_version=stripe_version, 

171 stripe_account=stripe_account, 

172 ) 

173 

174 first_id = self.data[0].id 

175 

176 params_with_filters = self._retrieve_params.copy() 

177 params_with_filters.update({"ending_before": first_id}) 

178 params_with_filters.update(params) 

179 

180 return self.list( 

181 api_key=api_key, 

182 stripe_version=stripe_version, 

183 stripe_account=stripe_account, 

184 **params_with_filters 

185 )