Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/stripe/stripe_object.py: 17%
176 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 datetime
4import json
5from copy import deepcopy
7import stripe
8from stripe import api_requestor, util, six
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 ""
21def _serialize_list(array, previous):
22 array = array or []
23 previous = previous or []
24 params = {}
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)
33 return params
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)
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__()
54 self._unsaved_values = set()
55 self._transient_values = set()
56 self._last_response = last_response
58 self._retrieve_params = params
59 self._previous = None
61 object.__setattr__(self, "api_key", api_key)
62 object.__setattr__(self, "stripe_version", stripe_version)
63 object.__setattr__(self, "stripe_account", stripe_account)
65 if id:
66 self["id"] = id
68 @property
69 def last_response(self):
70 return self._last_response
72 def update(self, update_dict):
73 for k in update_dict:
74 self._unsaved_values.add(k)
76 return super(StripeObject, self).update(update_dict)
78 def __setattr__(self, k, v):
79 if k[0] == "_" or k in self.__dict__:
80 return super(StripeObject, self).__setattr__(k, v)
82 self[k] = v
83 return None
85 def __getattr__(self, k):
86 if k[0] == "_":
87 raise AttributeError(k)
89 try:
90 return self[k]
91 except KeyError as err:
92 raise AttributeError(*err.args)
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]
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 )
110 # Allows for unpickling in Python 3.x
111 if not hasattr(self, "_unsaved_values"):
112 self._unsaved_values = set()
114 self._unsaved_values.add(k)
116 super(StripeObject, self).__setitem__(k, v)
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
134 def __delitem__(self, k):
135 super(StripeObject, self).__delitem__(k)
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)
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)
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
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
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 )
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()
219 self._transient_values = self._transient_values - set(values)
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 )
229 self._previous = values
231 @classmethod
232 def api_base(cls):
233 return None
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)
246 return util.convert_to_stripe_object(
247 response, api_key, self.stripe_version, self.stripe_account
248 )
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)
261 return response
263 def __repr__(self):
264 ident_parts = [type(self).__name__]
266 if isinstance(self.get("object"), six.string_types):
267 ident_parts.append(self.get("object"))
269 if isinstance(self.get("id"), six.string_types):
270 ident_parts.append("id=%s" % (self.get("id"),))
272 unicode_repr = "<%s at %s> JSON: %s" % (
273 " ".join(ident_parts),
274 hex(id(self)),
275 str(self),
276 )
278 if six.PY2:
279 return unicode_repr.encode("utf-8")
280 else:
281 return unicode_repr
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 )
291 def to_dict(self):
292 return dict(self)
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
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 }
310 @property
311 def stripe_id(self):
312 return self.id
314 def serialize(self, previous):
315 params = {}
316 unsaved_keys = self._unsaved_values or set()
317 previous = previous or self._previous or {}
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))
333 return params
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 )
348 copied._retrieve_params = self._retrieve_params
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)
355 return copied
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
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))
371 return copied