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

1from __future__ import absolute_import, division, print_function 

2 

3import datetime 

4import json 

5from copy import deepcopy 

6 

7import stripe 

8from stripe import api_requestor, util, six 

9 

10 

11def _compute_diff(current, previous): 

12 if isinstance(current, dict): 

13 previous = previous or {} 

14 diff = current.copy() 

15 for key in set(previous.keys()) - set(diff.keys()): 

16 diff[key] = "" 

17 return diff 

18 return current if current is not None else "" 

19 

20 

21def _serialize_list(array, previous): 

22 array = array or [] 

23 previous = previous or [] 

24 params = {} 

25 

26 for i, v in enumerate(array): 

27 previous_item = previous[i] if len(previous) > i else None 

28 if hasattr(v, "serialize"): 

29 params[str(i)] = v.serialize(previous_item) 

30 else: 

31 params[str(i)] = _compute_diff(v, previous_item) 

32 

33 return params 

34 

35 

36class StripeObject(dict): 

37 class ReprJSONEncoder(json.JSONEncoder): 

38 def default(self, obj): 

39 if isinstance(obj, datetime.datetime): 

40 return api_requestor._encode_datetime(obj) 

41 return super(StripeObject.ReprJSONEncoder, self).default(obj) 

42 

43 def __init__( 

44 self, 

45 id=None, 

46 api_key=None, 

47 stripe_version=None, 

48 stripe_account=None, 

49 last_response=None, 

50 **params 

51 ): 

52 super(StripeObject, self).__init__() 

53 

54 self._unsaved_values = set() 

55 self._transient_values = set() 

56 self._last_response = last_response 

57 

58 self._retrieve_params = params 

59 self._previous = None 

60 

61 object.__setattr__(self, "api_key", api_key) 

62 object.__setattr__(self, "stripe_version", stripe_version) 

63 object.__setattr__(self, "stripe_account", stripe_account) 

64 

65 if id: 

66 self["id"] = id 

67 

68 @property 

69 def last_response(self): 

70 return self._last_response 

71 

72 def update(self, update_dict): 

73 for k in update_dict: 

74 self._unsaved_values.add(k) 

75 

76 return super(StripeObject, self).update(update_dict) 

77 

78 def __setattr__(self, k, v): 

79 if k[0] == "_" or k in self.__dict__: 

80 return super(StripeObject, self).__setattr__(k, v) 

81 

82 self[k] = v 

83 return None 

84 

85 def __getattr__(self, k): 

86 if k[0] == "_": 

87 raise AttributeError(k) 

88 

89 try: 

90 return self[k] 

91 except KeyError as err: 

92 raise AttributeError(*err.args) 

93 

94 def __delattr__(self, k): 

95 if k[0] == "_" or k in self.__dict__: 

96 return super(StripeObject, self).__delattr__(k) 

97 else: 

98 del self[k] 

99 

100 def __setitem__(self, k, v): 

101 if v == "": 

102 raise ValueError( 

103 "You cannot set %s to an empty string on this object. " 

104 "The empty string is treated specially in our requests. " 

105 "If you'd like to delete the property using the save() method on this object, you may set %s.%s = None. " 

106 "Alternatively, you can pass %s='' to delete the property when using a resource method such as modify()." 

107 % (k, str(self), k, k) 

108 ) 

109 

110 # Allows for unpickling in Python 3.x 

111 if not hasattr(self, "_unsaved_values"): 

112 self._unsaved_values = set() 

113 

114 self._unsaved_values.add(k) 

115 

116 super(StripeObject, self).__setitem__(k, v) 

117 

118 def __getitem__(self, k): 

119 try: 

120 return super(StripeObject, self).__getitem__(k) 

121 except KeyError as err: 

122 if k in self._transient_values: 

123 raise KeyError( 

124 "%r. HINT: The %r attribute was set in the past." 

125 "It was then wiped when refreshing the object with " 

126 "the result returned by Stripe's API, probably as a " 

127 "result of a save(). The attributes currently " 

128 "available on this object are: %s" 

129 % (k, k, ", ".join(list(self.keys()))) 

130 ) 

131 else: 

132 raise err 

133 

134 def __delitem__(self, k): 

135 super(StripeObject, self).__delitem__(k) 

136 

137 # Allows for unpickling in Python 3.x 

138 if hasattr(self, "_unsaved_values") and k in self._unsaved_values: 

139 self._unsaved_values.remove(k) 

140 

141 # Custom unpickling method that uses `update` to update the dictionary 

142 # without calling __setitem__, which would fail if any value is an empty 

143 # string 

144 def __setstate__(self, state): 

145 self.update(state) 

146 

147 # Custom pickling method to ensure the instance is pickled as a custom 

148 # class and not as a dict, otherwise __setstate__ would not be called when 

149 # unpickling. 

150 def __reduce__(self): 

151 reduce_value = ( 

152 type(self), # callable 

153 ( # args 

154 self.get("id", None), 

155 self.api_key, 

156 self.stripe_version, 

157 self.stripe_account, 

158 ), 

159 dict(self), # state 

160 ) 

161 return reduce_value 

162 

163 @classmethod 

164 def construct_from( 

165 cls, 

166 values, 

167 key, 

168 stripe_version=None, 

169 stripe_account=None, 

170 last_response=None, 

171 ): 

172 instance = cls( 

173 values.get("id"), 

174 api_key=key, 

175 stripe_version=stripe_version, 

176 stripe_account=stripe_account, 

177 last_response=last_response, 

178 ) 

179 instance.refresh_from( 

180 values, 

181 api_key=key, 

182 stripe_version=stripe_version, 

183 stripe_account=stripe_account, 

184 last_response=last_response, 

185 ) 

186 return instance 

187 

188 def refresh_from( 

189 self, 

190 values, 

191 api_key=None, 

192 partial=False, 

193 stripe_version=None, 

194 stripe_account=None, 

195 last_response=None, 

196 ): 

197 self.api_key = api_key or getattr(values, "api_key", None) 

198 self.stripe_version = stripe_version or getattr( 

199 values, "stripe_version", None 

200 ) 

201 self.stripe_account = stripe_account or getattr( 

202 values, "stripe_account", None 

203 ) 

204 self._last_response = last_response or getattr( 

205 values, "_last_response", None 

206 ) 

207 

208 # Wipe old state before setting new. This is useful for e.g. 

209 # updating a customer, where there is no persistent card 

210 # parameter. Mark those values which don't persist as transient 

211 if partial: 

212 self._unsaved_values = self._unsaved_values - set(values) 

213 else: 

214 removed = set(self.keys()) - set(values) 

215 self._transient_values = self._transient_values | removed 

216 self._unsaved_values = set() 

217 self.clear() 

218 

219 self._transient_values = self._transient_values - set(values) 

220 

221 for k, v in six.iteritems(values): 

222 super(StripeObject, self).__setitem__( 

223 k, 

224 util.convert_to_stripe_object( 

225 v, api_key, stripe_version, stripe_account 

226 ), 

227 ) 

228 

229 self._previous = values 

230 

231 @classmethod 

232 def api_base(cls): 

233 return None 

234 

235 def request(self, method, url, params=None, headers=None): 

236 if params is None: 

237 params = self._retrieve_params 

238 requestor = api_requestor.APIRequestor( 

239 key=self.api_key, 

240 api_base=self.api_base(), 

241 api_version=self.stripe_version, 

242 account=self.stripe_account, 

243 ) 

244 response, api_key = requestor.request(method, url, params, headers) 

245 

246 return util.convert_to_stripe_object( 

247 response, api_key, self.stripe_version, self.stripe_account 

248 ) 

249 

250 def request_stream(self, method, url, params=None, headers=None): 

251 if params is None: 

252 params = self._retrieve_params 

253 requestor = api_requestor.APIRequestor( 

254 key=self.api_key, 

255 api_base=self.api_base(), 

256 api_version=self.stripe_version, 

257 account=self.stripe_account, 

258 ) 

259 response, _ = requestor.request_stream(method, url, params, headers) 

260 

261 return response 

262 

263 def __repr__(self): 

264 ident_parts = [type(self).__name__] 

265 

266 if isinstance(self.get("object"), six.string_types): 

267 ident_parts.append(self.get("object")) 

268 

269 if isinstance(self.get("id"), six.string_types): 

270 ident_parts.append("id=%s" % (self.get("id"),)) 

271 

272 unicode_repr = "<%s at %s> JSON: %s" % ( 

273 " ".join(ident_parts), 

274 hex(id(self)), 

275 str(self), 

276 ) 

277 

278 if six.PY2: 

279 return unicode_repr.encode("utf-8") 

280 else: 

281 return unicode_repr 

282 

283 def __str__(self): 

284 return json.dumps( 

285 self.to_dict_recursive(), 

286 sort_keys=True, 

287 indent=2, 

288 cls=self.ReprJSONEncoder, 

289 ) 

290 

291 def to_dict(self): 

292 return dict(self) 

293 

294 def to_dict_recursive(self): 

295 def maybe_to_dict_recursive(value): 

296 if value is None: 

297 return None 

298 elif isinstance(value, StripeObject): 

299 return value.to_dict_recursive() 

300 else: 

301 return value 

302 

303 return { 

304 key: list(map(maybe_to_dict_recursive, value)) 

305 if isinstance(value, list) 

306 else maybe_to_dict_recursive(value) 

307 for key, value in six.iteritems(dict(self)) 

308 } 

309 

310 @property 

311 def stripe_id(self): 

312 return self.id 

313 

314 def serialize(self, previous): 

315 params = {} 

316 unsaved_keys = self._unsaved_values or set() 

317 previous = previous or self._previous or {} 

318 

319 for k, v in six.iteritems(self): 

320 if k == "id" or (isinstance(k, str) and k.startswith("_")): 

321 continue 

322 elif isinstance(v, stripe.api_resources.abstract.APIResource): 

323 continue 

324 elif hasattr(v, "serialize"): 

325 child = v.serialize(previous.get(k, None)) 

326 if child != {}: 

327 params[k] = child 

328 elif k in unsaved_keys: 

329 params[k] = _compute_diff(v, previous.get(k, None)) 

330 elif k == "additional_owners" and v is not None: 

331 params[k] = _serialize_list(v, previous.get(k, None)) 

332 

333 return params 

334 

335 # This class overrides __setitem__ to throw exceptions on inputs that it 

336 # doesn't like. This can cause problems when we try to copy an object 

337 # wholesale because some data that's returned from the API may not be valid 

338 # if it was set to be set manually. Here we override the class' copy 

339 # arguments so that we can bypass these possible exceptions on __setitem__. 

340 def __copy__(self): 

341 copied = StripeObject( 

342 self.get("id"), 

343 self.api_key, 

344 stripe_version=self.stripe_version, 

345 stripe_account=self.stripe_account, 

346 ) 

347 

348 copied._retrieve_params = self._retrieve_params 

349 

350 for k, v in six.iteritems(self): 

351 # Call parent's __setitem__ to avoid checks that we've added in the 

352 # overridden version that can throw exceptions. 

353 super(StripeObject, copied).__setitem__(k, v) 

354 

355 return copied 

356 

357 # This class overrides __setitem__ to throw exceptions on inputs that it 

358 # doesn't like. This can cause problems when we try to copy an object 

359 # wholesale because some data that's returned from the API may not be valid 

360 # if it was set to be set manually. Here we override the class' copy 

361 # arguments so that we can bypass these possible exceptions on __setitem__. 

362 def __deepcopy__(self, memo): 

363 copied = self.__copy__() 

364 memo[id(self)] = copied 

365 

366 for k, v in six.iteritems(self): 

367 # Call parent's __setitem__ to avoid checks that we've added in the 

368 # overridden version that can throw exceptions. 

369 super(StripeObject, copied).__setitem__(k, deepcopy(v, memo)) 

370 

371 return copied