691 lines
22 KiB
Python
691 lines
22 KiB
Python
# pyright: strict
|
|
import json
|
|
from copy import deepcopy
|
|
from typing_extensions import TYPE_CHECKING, Type, Literal, Self, deprecated
|
|
from typing import (
|
|
Any,
|
|
Dict,
|
|
Generic,
|
|
List,
|
|
Optional,
|
|
Mapping,
|
|
Set,
|
|
Tuple,
|
|
ClassVar,
|
|
TypeVar,
|
|
Union,
|
|
cast,
|
|
overload,
|
|
)
|
|
|
|
# Used to break circular imports
|
|
import stripe # noqa: IMP101
|
|
from stripe import _util
|
|
|
|
from stripe._stripe_response import (
|
|
StripeResponse,
|
|
StripeStreamResponse,
|
|
StripeStreamResponseAsync,
|
|
)
|
|
from stripe._encode import (
|
|
_coerce_int64_string, # pyright: ignore[reportPrivateUsage]
|
|
_coerce_decimal_string, # pyright: ignore[reportPrivateUsage]
|
|
_make_suitable_for_json, # pyright: ignore[reportPrivateUsage]
|
|
)
|
|
from stripe._request_options import (
|
|
PERSISTENT_OPTIONS_KEYS,
|
|
extract_options_from_dict,
|
|
)
|
|
from stripe._api_mode import ApiMode
|
|
from stripe._base_address import BaseAddress
|
|
|
|
if TYPE_CHECKING:
|
|
from stripe._api_requestor import _APIRequestor # pyright: ignore[reportPrivateUsage]
|
|
|
|
|
|
@overload
|
|
def _compute_diff(
|
|
current: Dict[str, Any], previous: Optional[Dict[str, Any]]
|
|
) -> Dict[str, Any]: ...
|
|
|
|
|
|
@overload
|
|
def _compute_diff(
|
|
current: object, previous: Optional[Dict[str, Any]]
|
|
) -> object: ...
|
|
|
|
|
|
def _compute_diff(
|
|
current: object, previous: Optional[Dict[str, Any]]
|
|
) -> object:
|
|
if isinstance(current, dict):
|
|
current = cast(Dict[str, Any], current)
|
|
previous = previous or {}
|
|
diff = current.copy()
|
|
for key in set(previous.keys()) - set(diff.keys()):
|
|
diff[key] = ""
|
|
return diff
|
|
return current if current is not None else ""
|
|
|
|
|
|
def _serialize_list(
|
|
array: Optional[List[Any]], previous: List[Any]
|
|
) -> Dict[str, Any]:
|
|
array = array or []
|
|
previous = previous or []
|
|
params: Dict[str, Any] = {}
|
|
|
|
for i, v in enumerate(array):
|
|
previous_item = previous[i] if len(previous) > i else None
|
|
if hasattr(v, "serialize"):
|
|
params[str(i)] = v.serialize(previous_item)
|
|
else:
|
|
params[str(i)] = _compute_diff(v, previous_item)
|
|
|
|
return params
|
|
|
|
|
|
class StripeObject:
|
|
_retrieve_params: Mapping[str, Any]
|
|
_previous: Optional[Mapping[str, Any]]
|
|
|
|
def __init__(
|
|
self,
|
|
id: Optional[str] = None,
|
|
api_key: Optional[str] = None,
|
|
stripe_version: Optional[str] = None,
|
|
stripe_account: Optional[str] = None,
|
|
last_response: Optional[StripeResponse] = None,
|
|
*,
|
|
_requestor: Optional["_APIRequestor"] = None,
|
|
# TODO: is a more specific type possible here?
|
|
**params: Any,
|
|
):
|
|
self._data: Dict[str, Any] = {}
|
|
self._unsaved_values: Set[str] = set()
|
|
self._transient_values: Set[str] = set()
|
|
self._last_response = last_response
|
|
|
|
self._retrieve_params = params
|
|
self._previous = None
|
|
|
|
from stripe._api_requestor import _APIRequestor # pyright: ignore[reportPrivateUsage]
|
|
|
|
self._requestor = (
|
|
_APIRequestor._global_with_options( # pyright: ignore[reportPrivateUsage]
|
|
api_key=api_key,
|
|
stripe_version=stripe_version,
|
|
stripe_account=stripe_account,
|
|
)
|
|
if _requestor is None
|
|
else _requestor
|
|
)
|
|
|
|
if id:
|
|
self["id"] = id
|
|
|
|
@property
|
|
def api_key(self):
|
|
return self._requestor.api_key
|
|
|
|
@property
|
|
def stripe_account(self):
|
|
return self._requestor.stripe_account
|
|
|
|
@property
|
|
def stripe_version(self):
|
|
return self._requestor.stripe_version
|
|
|
|
@property
|
|
def last_response(self) -> Optional[StripeResponse]:
|
|
return self._last_response
|
|
|
|
def update(self, update_dict: Mapping[str, Any]) -> None:
|
|
for k in update_dict:
|
|
self._unsaved_values.add(k)
|
|
|
|
self._data.update(update_dict)
|
|
|
|
if not TYPE_CHECKING:
|
|
|
|
def __setattr__(self, k, v):
|
|
if k in PERSISTENT_OPTIONS_KEYS:
|
|
self._requestor = self._requestor._new_requestor_with_options(
|
|
{k: v}
|
|
)
|
|
return None
|
|
|
|
if k[0] == "_" or k in self.__dict__:
|
|
return super(StripeObject, self).__setattr__(k, v)
|
|
|
|
self[k] = v
|
|
return None
|
|
|
|
def __getattr__(self, k):
|
|
if k[0] == "_":
|
|
raise AttributeError(k)
|
|
|
|
try:
|
|
if k in self._field_remappings:
|
|
k = self._field_remappings[k]
|
|
return self[k]
|
|
except KeyError as err:
|
|
raise AttributeError(*err.args) from err
|
|
|
|
def __delattr__(self, k):
|
|
if k[0] == "_" or k in self.__dict__:
|
|
return super(StripeObject, self).__delattr__(k)
|
|
else:
|
|
del self[k]
|
|
|
|
def __setitem__(self, k: str, v: Any) -> None:
|
|
if v == "":
|
|
raise ValueError(
|
|
"You cannot set %s to an empty string on this object. "
|
|
"The empty string is treated specially in our requests. "
|
|
"If you'd like to delete the property using the save() method on this object, you may set %s.%s=None. "
|
|
"Alternatively, you can pass %s='' to delete the property when using a resource method such as modify()."
|
|
% (k, str(self), k, k)
|
|
)
|
|
|
|
# Allows for unpickling in Python 3.x
|
|
if not hasattr(self, "_unsaved_values"):
|
|
self._unsaved_values = set()
|
|
if not hasattr(self, "_data"):
|
|
self._data = {}
|
|
|
|
self._unsaved_values.add(k)
|
|
|
|
self._data[k] = v
|
|
|
|
def __getitem__(self, k: str) -> Any:
|
|
try:
|
|
return self._data[k]
|
|
except KeyError as err:
|
|
if k in self._transient_values:
|
|
raise KeyError(
|
|
"%r. HINT: The %r attribute was set in the past."
|
|
"It was then wiped when refreshing the object with "
|
|
"the result returned by Stripe's API, probably as a "
|
|
"result of a save(). The attributes currently "
|
|
"available on this object are: %s"
|
|
% (k, k, ", ".join(list(self._data.keys())))
|
|
)
|
|
else:
|
|
from stripe._invoice import Invoice
|
|
|
|
# super specific one-off case to help users debug this property disappearing
|
|
# see also: https://go/j/DEVSDK-2835
|
|
if isinstance(self, Invoice) and k == "payment_intent":
|
|
raise KeyError(
|
|
"The 'payment_intent' attribute is no longer available on Invoice objects. See the docs for more details: https://docs.stripe.com/changelog/basil/2025-03-31/add-support-for-multiple-partial-payments-on-invoices#why-is-this-a-breaking-change"
|
|
)
|
|
|
|
raise err
|
|
|
|
def __delitem__(self, k: str) -> None:
|
|
del self._data[k]
|
|
|
|
# Allows for unpickling in Python 3.x
|
|
if hasattr(self, "_unsaved_values") and k in self._unsaved_values:
|
|
self._unsaved_values.remove(k)
|
|
|
|
def __contains__(self, k: object) -> bool:
|
|
return k in self._data
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
if isinstance(other, StripeObject):
|
|
return type(self) is type(other) and self._data == other._data
|
|
return NotImplemented
|
|
|
|
# Custom unpickling method that updates _data directly
|
|
# without calling __setitem__, which would fail if any value is an empty
|
|
# string
|
|
def __setstate__(self, state: Dict[str, Any]) -> None:
|
|
self._data.update(state)
|
|
self._unsaved_values.update(state.keys())
|
|
|
|
# Custom pickling method to ensure the instance is pickled as a custom
|
|
# class and not as a dict, otherwise __setstate__ would not be called when
|
|
# unpickling.
|
|
def __reduce__(self) -> Tuple[Any, ...]:
|
|
reduce_value = (
|
|
type(self), # callable
|
|
( # args
|
|
getattr(self, "id", None),
|
|
self.api_key,
|
|
self.stripe_version,
|
|
self.stripe_account,
|
|
),
|
|
self._data.copy(), # state
|
|
)
|
|
return reduce_value
|
|
|
|
@classmethod
|
|
def construct_from(
|
|
cls,
|
|
values: Dict[str, Any],
|
|
key: Optional[str],
|
|
stripe_version: Optional[str] = None,
|
|
stripe_account: Optional[str] = None,
|
|
last_response: Optional[StripeResponse] = None,
|
|
*,
|
|
api_mode: ApiMode = "V1",
|
|
) -> Self:
|
|
from stripe._api_requestor import _APIRequestor # pyright: ignore[reportPrivateUsage]
|
|
|
|
return cls._construct_from(
|
|
values=values,
|
|
requestor=_APIRequestor._global_with_options( # pyright: ignore[reportPrivateUsage]
|
|
api_key=key,
|
|
stripe_version=stripe_version,
|
|
stripe_account=stripe_account,
|
|
),
|
|
api_mode=api_mode,
|
|
last_response=last_response,
|
|
)
|
|
|
|
@classmethod
|
|
def _construct_from(
|
|
cls,
|
|
*,
|
|
values: "Union[Dict[str, Any], StripeObject]",
|
|
last_response: Optional[StripeResponse] = None,
|
|
requestor: "_APIRequestor",
|
|
api_mode: ApiMode,
|
|
) -> Self:
|
|
raw = values._data if isinstance(values, StripeObject) else values
|
|
instance = cls(
|
|
raw.get("id"),
|
|
last_response=last_response,
|
|
_requestor=requestor,
|
|
)
|
|
instance._refresh_from(
|
|
values=values,
|
|
last_response=last_response,
|
|
requestor=requestor,
|
|
api_mode=api_mode,
|
|
)
|
|
return instance
|
|
|
|
def refresh_from(
|
|
self,
|
|
values: Dict[str, Any],
|
|
api_key: Optional[str] = None,
|
|
partial: Optional[bool] = False,
|
|
stripe_version: Optional[str] = None,
|
|
stripe_account: Optional[str] = None,
|
|
last_response: Optional[StripeResponse] = None,
|
|
*,
|
|
api_mode: ApiMode = "V1",
|
|
) -> None:
|
|
self._refresh_from(
|
|
values=values,
|
|
partial=partial,
|
|
last_response=last_response,
|
|
requestor=self._requestor._new_requestor_with_options( # pyright: ignore[reportPrivateUsage]
|
|
{
|
|
"api_key": api_key,
|
|
"stripe_version": stripe_version,
|
|
"stripe_account": stripe_account,
|
|
}
|
|
),
|
|
api_mode=api_mode,
|
|
)
|
|
|
|
def _refresh_from(
|
|
self,
|
|
*,
|
|
values: "Union[Dict[str, Any], StripeObject]",
|
|
partial: Optional[bool] = False,
|
|
last_response: Optional[StripeResponse] = None,
|
|
requestor: Optional["_APIRequestor"] = None,
|
|
api_mode: ApiMode,
|
|
) -> None:
|
|
self._requestor = requestor or self._requestor
|
|
self._last_response = last_response or getattr(
|
|
values, "_last_response", None
|
|
)
|
|
|
|
# When called from APIResource._request, values may be a StripeObject
|
|
if isinstance(values, StripeObject):
|
|
values = values._data
|
|
|
|
# Wipe old state before setting new. This is useful for e.g.
|
|
# updating a customer, where there is no persistent card
|
|
# parameter. Mark those values which don't persist as transient
|
|
if partial:
|
|
self._unsaved_values = self._unsaved_values - set(values)
|
|
else:
|
|
removed = set(self._data.keys()) - set(values)
|
|
self._transient_values = self._transient_values | removed
|
|
self._unsaved_values = set()
|
|
self._data.clear()
|
|
|
|
self._transient_values = self._transient_values - set(values)
|
|
|
|
for k, v in values.items():
|
|
# Apply field encoding coercion (e.g. int64_string: str → int)
|
|
v = self._coerce_field_value(k, v)
|
|
inner_class = self._get_inner_class_type(k)
|
|
is_dict = self._get_inner_class_is_beneath_dict(k)
|
|
if is_dict:
|
|
obj = {
|
|
k: None
|
|
if v is None
|
|
else cast(
|
|
StripeObject,
|
|
_util._convert_to_stripe_object( # pyright: ignore[reportPrivateUsage]
|
|
resp=v,
|
|
params=None,
|
|
klass_=inner_class,
|
|
requestor=self._requestor,
|
|
api_mode=api_mode,
|
|
),
|
|
)
|
|
for k, v in v.items()
|
|
}
|
|
else:
|
|
obj = cast(
|
|
Union[StripeObject, List[StripeObject]],
|
|
_util._convert_to_stripe_object( # pyright: ignore[reportPrivateUsage]
|
|
resp=v,
|
|
params=None,
|
|
klass_=inner_class,
|
|
requestor=self._requestor,
|
|
api_mode=api_mode,
|
|
),
|
|
)
|
|
self._data[k] = obj
|
|
|
|
self._previous = values
|
|
|
|
@deprecated("This will be removed in a future version of stripe-python.")
|
|
def request(
|
|
self,
|
|
method: Literal["get", "post", "delete"],
|
|
url: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
base_address: BaseAddress = "api",
|
|
) -> "StripeObject":
|
|
return StripeObject._request(
|
|
self,
|
|
method,
|
|
url,
|
|
params=params,
|
|
base_address=base_address,
|
|
)
|
|
|
|
def _request(
|
|
self,
|
|
method: Literal["get", "post", "delete"],
|
|
url: str,
|
|
params: Optional[Mapping[str, Any]] = None,
|
|
usage: Optional[List[str]] = None,
|
|
*,
|
|
base_address: BaseAddress,
|
|
) -> "StripeObject":
|
|
if params is None:
|
|
params = self._retrieve_params
|
|
|
|
request_options, request_params = extract_options_from_dict(params)
|
|
|
|
return self._requestor.request(
|
|
method,
|
|
url,
|
|
params=request_params,
|
|
options=request_options,
|
|
base_address=base_address,
|
|
usage=usage,
|
|
)
|
|
|
|
async def _request_async(
|
|
self,
|
|
method: Literal["get", "post", "delete"],
|
|
url: str,
|
|
params: Optional[Mapping[str, Any]] = None,
|
|
usage: Optional[List[str]] = None,
|
|
*,
|
|
base_address: BaseAddress,
|
|
) -> "StripeObject":
|
|
if params is None:
|
|
params = self._retrieve_params
|
|
|
|
request_options, request_params = extract_options_from_dict(params)
|
|
|
|
return await self._requestor.request_async(
|
|
method,
|
|
url,
|
|
params=request_params,
|
|
options=request_options,
|
|
base_address=base_address,
|
|
usage=usage,
|
|
)
|
|
|
|
def _request_stream(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
params: Optional[Mapping[str, Any]] = None,
|
|
*,
|
|
base_address: BaseAddress = "api",
|
|
) -> StripeStreamResponse:
|
|
if params is None:
|
|
params = self._retrieve_params
|
|
|
|
request_options, request_params = extract_options_from_dict(params)
|
|
return self._requestor.request_stream(
|
|
method,
|
|
url,
|
|
params=request_params,
|
|
options=request_options,
|
|
base_address=base_address,
|
|
)
|
|
|
|
async def _request_stream_async(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
params: Optional[Mapping[str, Any]] = None,
|
|
*,
|
|
base_address: BaseAddress = "api",
|
|
) -> StripeStreamResponseAsync:
|
|
if params is None:
|
|
params = self._retrieve_params
|
|
|
|
request_options, request_params = extract_options_from_dict(params)
|
|
return await self._requestor.request_stream_async(
|
|
method,
|
|
url,
|
|
params=request_params,
|
|
options=request_options,
|
|
base_address=base_address,
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
ident_parts = [type(self).__name__]
|
|
|
|
obj_str = getattr(self, "object", None)
|
|
if isinstance(obj_str, str):
|
|
ident_parts.append(obj_str)
|
|
|
|
obj_id = getattr(self, "id", None)
|
|
if isinstance(obj_id, str):
|
|
ident_parts.append("id=%s" % (obj_id,))
|
|
|
|
unicode_repr = "<%s at %s> JSON: %s" % (
|
|
" ".join(ident_parts),
|
|
hex(id(self)),
|
|
str(self),
|
|
)
|
|
return unicode_repr
|
|
|
|
def __str__(self) -> str:
|
|
return json.dumps(
|
|
self._to_dict_recursive(),
|
|
sort_keys=True,
|
|
indent=2,
|
|
default=_make_suitable_for_json,
|
|
)
|
|
|
|
def to_dict(
|
|
self, recursive: bool = True, for_json: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Dump the object's backing data. Recurses by default, but you can opt-out of that behavior by passing `recursive=False`.
|
|
Pass `for_json=True` to convert non-JSON-serializable values (e.g. Decimal -> str)
|
|
"""
|
|
if recursive:
|
|
return self._to_dict_recursive(for_json=for_json)
|
|
|
|
# shallow copy, so nested objects will be shared
|
|
return self._data.copy()
|
|
|
|
def _to_dict_recursive(self, for_json: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
used by __str__ to serialize the whole object
|
|
"""
|
|
|
|
def maybe_to_dict_recursive(
|
|
value: Optional[Union[StripeObject, Dict[str, Any]]],
|
|
) -> Optional[Dict[str, Any]]:
|
|
if value is None:
|
|
return None
|
|
elif isinstance(value, StripeObject):
|
|
return value._to_dict_recursive(for_json=for_json)
|
|
elif for_json:
|
|
return _make_suitable_for_json(value)
|
|
else:
|
|
return value
|
|
|
|
return {
|
|
key: list(map(maybe_to_dict_recursive, cast(List[Any], value)))
|
|
if isinstance(value, list)
|
|
else maybe_to_dict_recursive(value)
|
|
for key, value in self._data.items()
|
|
}
|
|
|
|
def serialize(
|
|
self, previous: Optional[Mapping[str, Any]]
|
|
) -> Dict[str, Any]:
|
|
params: Dict[str, Any] = {}
|
|
unsaved_keys = self._unsaved_values or set()
|
|
previous_raw = previous or self._previous or {}
|
|
if isinstance(previous_raw, StripeObject):
|
|
previous_raw = previous_raw._data
|
|
prev: Dict[str, Any] = dict(previous_raw)
|
|
|
|
for k, v in self._data.items():
|
|
if k == "id" or k.startswith("_"):
|
|
continue
|
|
elif isinstance(v, stripe.APIResource):
|
|
continue
|
|
elif hasattr(v, "serialize"):
|
|
child = v.serialize(prev.get(k, None))
|
|
if child != {}:
|
|
params[k] = child
|
|
elif k in unsaved_keys:
|
|
params[k] = _compute_diff(v, prev.get(k, None))
|
|
elif k == "additional_owners" and v is not None:
|
|
params[k] = _serialize_list(v, prev.get(k, None))
|
|
|
|
return params
|
|
|
|
# This class overrides __setitem__ to throw exceptions on inputs that it
|
|
# doesn't like. This can cause problems when we try to copy an object
|
|
# wholesale because some data that's returned from the API may not be valid
|
|
# if it was set to be set manually. Here we override the class' copy
|
|
# arguments so that we can bypass these possible exceptions on __setitem__.
|
|
def __copy__(self) -> "StripeObject":
|
|
copied = StripeObject(
|
|
getattr(self, "id", None),
|
|
self.api_key,
|
|
stripe_version=self.stripe_version,
|
|
stripe_account=self.stripe_account,
|
|
)
|
|
|
|
copied._retrieve_params = self._retrieve_params
|
|
|
|
for k, v in self._data.items():
|
|
# Write to _data directly to avoid checks that we've added in the
|
|
# overridden __setitem__ that can throw exceptions.
|
|
copied._data[k] = v
|
|
|
|
return copied
|
|
|
|
# This class overrides __setitem__ to throw exceptions on inputs that it
|
|
# doesn't like. This can cause problems when we try to copy an object
|
|
# wholesale because some data that's returned from the API may not be valid
|
|
# if it was set to be set manually. Here we override the class' copy
|
|
# arguments so that we can bypass these possible exceptions on __setitem__.
|
|
def __deepcopy__(self, memo: Dict[int, Any]) -> "StripeObject":
|
|
copied = self.__copy__()
|
|
memo[id(self)] = copied
|
|
|
|
for k, v in self._data.items():
|
|
# Write to _data directly to avoid checks that we've added in the
|
|
# overridden __setitem__ that can throw exceptions.
|
|
copied._data[k] = deepcopy(v, memo)
|
|
|
|
return copied
|
|
|
|
_field_remappings: ClassVar[Dict[str, str]] = {}
|
|
|
|
_inner_class_types: ClassVar[Dict[str, Type["StripeObject"]]] = {}
|
|
_inner_class_dicts: ClassVar[List[str]] = []
|
|
_field_encodings: ClassVar[Dict[str, str]] = {}
|
|
|
|
def _get_inner_class_type(
|
|
self, field_name: str
|
|
) -> Optional[Type["StripeObject"]]:
|
|
return self._inner_class_types.get(field_name)
|
|
|
|
def _get_inner_class_is_beneath_dict(self, field_name: str):
|
|
return field_name in self._inner_class_dicts
|
|
|
|
def _coerce_field_value(self, field_name: str, value: Any) -> Any:
|
|
"""
|
|
Convert JSON types to more applicable Python types, if able.
|
|
|
|
For example, "int64_string"s become `int`s.
|
|
"""
|
|
|
|
# WARNING: if you edit this function to produce a type that's not json-serializable, you need to update `_make_suitable_for_json` as well.
|
|
# By default, Python will only correctly dump a few standard types, so we have to handle the rest
|
|
|
|
encoding = self._field_encodings.get(field_name)
|
|
if encoding is None or value is None:
|
|
return value
|
|
|
|
if encoding == "int64_string":
|
|
return _coerce_int64_string(value, encode=False)
|
|
|
|
if encoding == "decimal_string":
|
|
return _coerce_decimal_string(value, encode=False)
|
|
|
|
return value
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class UntypedStripeObject(StripeObject, Generic[T]):
|
|
"""
|
|
A normal StripeObject, but it exposes `__getattr__`/`__setattr__` instead of hiding them, effectively removing type information.
|
|
|
|
Because metadata & similar are supposed to be an untyped `dict`, we don't want to show type errors for arbitrary key access.
|
|
|
|
Is generic on its value type
|
|
"""
|
|
|
|
def __init__(*args, **kwargs: Any):
|
|
raise ValueError("this is not for runtime use, just typing")
|
|
|
|
# This class is never actually used at runtime, it's just here for typechecking reasons
|
|
def __setattr__(self, k: str, v: T): ...
|
|
|
|
def __getattr__(self, k: str) -> T: ...
|
|
|
|
def __delattr__(self, k: str): ...
|