7175125064

Committed 12 Dec 2023 12:23AM UTC coverage: 96.171% (+0.07%) from 96.103%

Build # 7175125064

Build Type

push

github

Committed by

web-flow

Commit Message

Mark defunct and internal methods as deprecated (#1101)

Run Details

852 of 1290 branches covered (0.0%)

Branch coverage included in aggregate %.

72 of 76 new or added lines in 8 files covered. (94.74%)

1 existing line in 1 file now uncovered.

33710 of 34648 relevant lines covered (97.29%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.18

/stripe/_stripe_object.py

# pyright: strict
import datetime
import json
from copy import deepcopy
from typing_extensions import TYPE_CHECKING, Type, Literal, Self
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Mapping,
    Set,
    Tuple,
    ClassVar,
    Union,
    cast,
    overload,
)

# Used to break circular imports
import stripe  # noqa: IMP101
from stripe._encode import _encode_datetime  # pyright: ignore
from stripe import _util

from stripe._stripe_response import StripeResponse, StripeStreamResponse
import warnings


@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(Dict[str, Any]):
    class _ReprJSONEncoder(json.JSONEncoder):
        def default(self, o: Any) -> Any:
            if isinstance(o, datetime.datetime):
                # pyright complains that _encode_datetime is "private", but it's
                # private to outsiders, not to stripe_object
                return _encode_datetime(o)
            return super(StripeObject._ReprJSONEncoder, self).default(o)

    @_util.deprecated(
        "For internal stripe-python use only. The public interface will be removed in a future version"
    )
    class ReprJSONEncoder(_ReprJSONEncoder):
        pass

    _retrieve_params: Dict[str, Any]
    _previous: Optional[Dict[str, Any]]

    api_key: Optional[str]
    stripe_version: Optional[str]
    stripe_account: Optional[str]

    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,
        # TODO: is a more specific type possible here?
        **params: Any
    ):
        super(StripeObject, self).__init__()

        self._unsaved_values: Set[str] = set()
        self._transient_values: Set[str] = set()
        self._last_response = last_response

        self._retrieve_params = params
        self._previous = None

        object.__setattr__(self, "api_key", api_key)
        object.__setattr__(self, "stripe_version", stripe_version)
        object.__setattr__(self, "stripe_account", stripe_account)

        if id:
            self["id"] = id

    @property
    def last_response(self) -> Optional[StripeResponse]:
        return self._last_response

    # StripeObject inherits from `dict` which has an update method, and this doesn't quite match
    # the full signature of the update method in MutableMapping. But we ignore.
    def update(  # pyright: ignore
        self, update_dict: Mapping[str, Any]
    ) -> None:
        for k in update_dict:
            self._unsaved_values.add(k)

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

    if not TYPE_CHECKING:

        def __setattr__(self, k, v):
            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:
                return self[k]
            except KeyError as err:
                raise AttributeError(*err.args)

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

        self._unsaved_values.add(k)

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

    def __getitem__(self, k: str) -> Any:
        try:
            return super(StripeObject, self).__getitem__(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.keys())))
                )
            else:
                raise err

    def __delitem__(self, k: str) -> None:
        super(StripeObject, self).__delitem__(k)

        # Allows for unpickling in Python 3.x
        if hasattr(self, "_unsaved_values") and k in self._unsaved_values:
            self._unsaved_values.remove(k)

    # Custom unpickling method that uses `update` to update the dictionary
    # without calling __setitem__, which would fail if any value is an empty
    # string
    def __setstate__(self, state: Dict[str, Any]) -> None:
        self.update(state)

    # 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
                self.get("id", None),
                self.api_key,
                self.stripe_version,
                self.stripe_account,
            ),
            dict(self),  # 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,
    ) -> Self:
        instance = cls(
            values.get("id"),
            api_key=key,
            stripe_version=stripe_version,
            stripe_account=stripe_account,
            last_response=last_response,
        )
        instance.refresh_from(
            values,
            api_key=key,
            stripe_version=stripe_version,
            stripe_account=stripe_account,
            last_response=last_response,
        )
        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,
    ) -> None:
        self.api_key = api_key or getattr(values, "api_key", None)
        self.stripe_version = stripe_version or getattr(
            values, "stripe_version", None
        )
        self.stripe_account = stripe_account or getattr(
            values, "stripe_account", None
        )
        self._last_response = last_response or getattr(
            values, "_last_response", None
        )

        # 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.keys()) - set(values)
            self._transient_values = self._transient_values | removed
            self._unsaved_values = set()
            self.clear()

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

        for k, v in values.items():
            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(
                            v,
                            api_key,
                            stripe_version,
                            stripe_account,
                            None,
                            inner_class,
                        ),
                    )
                    for k, v in v.items()
                }
            else:
                obj = cast(
                    Union[StripeObject, List[StripeObject]],
                    _util.convert_to_stripe_object(
                        v,
                        api_key,
                        stripe_version,
                        stripe_account,
                        None,
                        inner_class,
                    ),
                )
            super(StripeObject, self).__setitem__(k, obj)

        self._previous = values

    @classmethod
    @_util.deprecated(
        "This will be removed in a future version of stripe-python."
    )
    def api_base(cls) -> Optional[str]:
        return None

    def request(
        self,
        method: Literal["get", "post", "delete"],
        url: str,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> "StripeObject":
        return StripeObject._request(
            self, method, url, headers=headers, params=params
        )

    # The `method_` and `url_` arguments are suffixed with an underscore to
    # avoid conflicting with actual request parameters in `params`.
    def _request(
        self,
        method_: Literal["get", "post", "delete"],
        url_: str,
        api_key: Optional[str] = None,
        idempotency_key: Optional[str] = None,
        stripe_version: Optional[str] = None,
        stripe_account: Optional[str] = None,
        headers: Optional[Dict[str, str]] = None,
        params: Optional[Mapping[str, Any]] = None,
    ) -> "StripeObject":
        params = None if params is None else dict(params)
        api_key = _util.read_special_variable(params, "api_key", api_key)
        idempotency_key = _util.read_special_variable(
            params, "idempotency_key", idempotency_key
        )
        stripe_version = _util.read_special_variable(
            params, "stripe_version", stripe_version
        )
        stripe_account = _util.read_special_variable(
            params, "stripe_account", stripe_account
        )
        headers = _util.read_special_variable(params, "headers", headers)

        stripe_account = stripe_account or self.stripe_account
        stripe_version = stripe_version or self.stripe_version
        api_key = api_key or self.api_key
        params = params or self._retrieve_params
        api_base = None
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            api_base = self.api_base()  # pyright: ignore[reportDeprecated]

        requestor = stripe.APIRequestor(
            key=api_key,
            api_base=api_base,
            api_version=stripe_version,
            account=stripe_account,
        )

        if idempotency_key is not None:
            headers = {} if headers is None else headers.copy()
            headers.update(_util.populate_headers(idempotency_key))

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

        return _util.convert_to_stripe_object(
            response, api_key, stripe_version, stripe_account, params
        )

    def request_stream(
        self,
        method: str,
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        headers: Optional[Mapping[str, str]] = None,
    ) -> StripeStreamResponse:
        if params is None:
            params = self._retrieve_params
        api_base = None
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            api_base = self.api_base()  # pyright: ignore[reportDeprecated]
        requestor = stripe.APIRequestor(
            key=self.api_key,
            api_base=api_base,
            api_version=self.stripe_version,
            account=self.stripe_account,
        )
        response, _ = requestor.request_stream(method, url, params, headers)

        return response

    def __repr__(self) -> str:
        ident_parts = [type(self).__name__]

        obj_str = self.get("object")
        if isinstance(obj_str, str):
            ident_parts.append(obj_str)

        if isinstance(self.get("id"), str):
            ident_parts.append("id=%s" % (self.get("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,
            cls=self._ReprJSONEncoder,
        )

    @_util.deprecated(
        "Deprecated. The public interface will be removed in a future version."
    )
    def to_dict(self) -> Dict[str, Any]:
        return dict(self)

    def _to_dict_recursive(self) -> Dict[str, Any]:
        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()
            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 dict(self).items()
        }

    @_util.deprecated(
        "For internal stripe-python use only. The public interface will be removed in a future version."
    )
    def to_dict_recursive(self) -> Dict[str, Any]:
        return self._to_dict_recursive()

    @property
    @_util.deprecated(
        "For internal stripe-python use only. The public interface will be removed in a future version."
    )
    def stripe_id(self) -> Optional[str]:
        return getattr(self, "id")

    def serialize(self, previous: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        unsaved_keys = self._unsaved_values or set()
        previous = previous or self._previous or {}

        for k, v in self.items():
            if k == "id" or k.startswith("_"):
                continue
            elif isinstance(v, stripe.APIResource):
                continue
            elif hasattr(v, "serialize"):
                child = v.serialize(previous.get(k, None))
                if child != {}:
                    params[k] = child
            elif k in unsaved_keys:
                params[k] = _compute_diff(v, previous.get(k, None))
            elif k == "additional_owners" and v is not None:
                params[k] = _serialize_list(v, previous.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(
            self.get("id"),
            self.api_key,
            stripe_version=self.stripe_version,
            stripe_account=self.stripe_account,
        )

        copied._retrieve_params = self._retrieve_params

        for k, v in self.items():
            # Call parent's __setitem__ to avoid checks that we've added in the
            # overridden version that can throw exceptions.
            super(StripeObject, copied).__setitem__(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.items():
            # Call parent's __setitem__ to avoid checks that we've added in the
            # overridden version that can throw exceptions.
            super(StripeObject, copied).__setitem__(k, deepcopy(v, memo))

        return copied

    _inner_class_types: ClassVar[Dict[str, Type["StripeObject"]]] = {}
    _inner_class_dicts: ClassVar[List[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

1	# pyright: strict
2	import datetime	1✔
3	import json	1✔
4	from copy import deepcopy	1✔
5	from typing_extensions import TYPE_CHECKING, Type, Literal, Self	1✔
6	from typing import (	1✔
7	Any,
8	Dict,
9	List,
10	Optional,
11	Mapping,
12	Set,
13	Tuple,
14	ClassVar,
15	Union,
16	cast,
17	overload,
18	)
19
20	# Used to break circular imports
21	import stripe # noqa: IMP101	1✔
22	from stripe._encode import _encode_datetime # pyright: ignore	1✔
23	from stripe import _util	1✔
24
25	from stripe._stripe_response import StripeResponse, StripeStreamResponse	1✔
26	import warnings	1✔
27
28
29	@overload	1✔
30	def _compute_diff(	1✔
31	current: Dict[str, Any], previous: Optional[Dict[str, Any]]
32	) -> Dict[str, Any]:
33	...	×
34
35
36	@overload	1✔
37	def _compute_diff(	1✔
38	current: object, previous: Optional[Dict[str, Any]]
39	) -> object:
40	...	×
41
42
43	def _compute_diff(	1✔
44	current: object, previous: Optional[Dict[str, Any]]
45	) -> object:
46	if isinstance(current, dict):	1✔
47	current = cast(Dict[str, Any], current)	1✔
48	previous = previous or {}	1✔
49	diff = current.copy()	1✔
50	for key in set(previous.keys()) - set(diff.keys()):	1✔
51	diff[key] = ""	1✔
52	return diff	1✔
53	return current if current is not None else ""	1✔
54
55
56	def _serialize_list(	1✔
57	array: Optional[List[Any]], previous: List[Any]
58	) -> Dict[str, Any]:
59	array = array or []	1✔
60	previous = previous or []	1✔
61	params: Dict[str, Any] = {}	1✔
62
63	for i, v in enumerate(array):	1✔
64	previous_item = previous[i] if len(previous) > i else None	1✔
65	if hasattr(v, "serialize"):	1✔
66	params[str(i)] = v.serialize(previous_item)	1✔
67	else:
68	params[str(i)] = _compute_diff(v, previous_item)	1✔
69
70	return params	1✔
71
72
73	class StripeObject(Dict[str, Any]):	1✔
74	class _ReprJSONEncoder(json.JSONEncoder):	1✔
75	def default(self, o: Any) -> Any:	1✔
76	if isinstance(o, datetime.datetime):	1!
77	# pyright complains that _encode_datetime is "private", but it's
78	# private to outsiders, not to stripe_object
79	return _encode_datetime(o)	1✔
NEW 80	return super(StripeObject._ReprJSONEncoder, self).default(o)	×
81
82	@_util.deprecated(	1✔
83	"For internal stripe-python use only. The public interface will be removed in a future version"
84	)
85	class ReprJSONEncoder(_ReprJSONEncoder):	1✔
86	pass	1✔
87
88	_retrieve_params: Dict[str, Any]	1✔
89	_previous: Optional[Dict[str, Any]]	1✔
90
91	api_key: Optional[str]	1✔
92	stripe_version: Optional[str]	1✔
93	stripe_account: Optional[str]	1✔
94
95	def __init__(	1✔
96	self,
97	id: Optional[str] = None,
98	api_key: Optional[str] = None,
99	stripe_version: Optional[str] = None,
100	stripe_account: Optional[str] = None,
101	last_response: Optional[StripeResponse] = None,
102	# TODO: is a more specific type possible here?
103	**params: Any
104	):
105	super(StripeObject, self).__init__()	1✔
106
107	self._unsaved_values: Set[str] = set()	1✔
108	self._transient_values: Set[str] = set()	1✔
109	self._last_response = last_response	1✔
110
111	self._retrieve_params = params	1✔
112	self._previous = None	1✔
113
114	object.__setattr__(self, "api_key", api_key)	1✔
115	object.__setattr__(self, "stripe_version", stripe_version)	1✔
116	object.__setattr__(self, "stripe_account", stripe_account)	1✔
117
118	if id:	1✔
119	self["id"] = id	1✔
120
121	@property	1✔
122	def last_response(self) -> Optional[StripeResponse]:	1✔
123	return self._last_response	1✔
124
125	# StripeObject inherits from `dict` which has an update method, and this doesn't quite match
126	# the full signature of the update method in MutableMapping. But we ignore.
127	def update( # pyright: ignore	1✔
128	self, update_dict: Mapping[str, Any]
129	) -> None:
130	for k in update_dict:	1✔
131	self._unsaved_values.add(k)	1✔
132
133	return super(StripeObject, self).update(update_dict)	1✔
134
135	if not TYPE_CHECKING:	1!
136
137	def __setattr__(self, k, v):	1✔
138	if k[0] == "_" or k in self.__dict__:	1✔
139	return super(StripeObject, self).__setattr__(k, v)	1✔
140
141	self[k] = v	1✔
142	return None	1✔
143
144	def __getattr__(self, k):	1✔
145	if k[0] == "_":	1!
146	raise AttributeError(k)	×
147
148	try:	1✔
149	return self[k]	1✔
150	except KeyError as err:	1✔
151	raise AttributeError(*err.args)	1✔
152
153	def __delattr__(self, k):	1✔
154	if k[0] == "_" or k in self.__dict__:	1!
155	return super(StripeObject, self).__delattr__(k)	×
156	else:
157	del self[k]	1✔
158
159	def __setitem__(self, k: str, v: Any) -> None:	1✔
160	if v == "":	1✔
161	raise ValueError(	1✔
162	"You cannot set %s to an empty string on this object. "
163	"The empty string is treated specially in our requests. "
164	"If you'd like to delete the property using the save() method on this object, you may set %s.%s=None. "
165	"Alternatively, you can pass %s='' to delete the property when using a resource method such as modify()."
166	% (k, str(self), k, k)
167	)
168
169	# Allows for unpickling in Python 3.x
170	if not hasattr(self, "_unsaved_values"):	1!
171	self._unsaved_values = set()	×
172
173	self._unsaved_values.add(k)	1✔
174
175	super(StripeObject, self).__setitem__(k, v)	1✔
176
177	def __getitem__(self, k: str) -> Any:	1✔
178	try:	1✔
179	return super(StripeObject, self).__getitem__(k)	1✔
180	except KeyError as err:	1✔
181	if k in self._transient_values:	1✔
182	raise KeyError(	1✔
183	"%r. HINT: The %r attribute was set in the past."
184	"It was then wiped when refreshing the object with "
185	"the result returned by Stripe's API, probably as a "
186	"result of a save(). The attributes currently "
187	"available on this object are: %s"
188	% (k, k, ", ".join(list(self.keys())))
189	)
190	else:
191	raise err	1✔
192
193	def __delitem__(self, k: str) -> None:	1✔
194	super(StripeObject, self).__delitem__(k)	1✔
195
196	# Allows for unpickling in Python 3.x
197	if hasattr(self, "_unsaved_values") and k in self._unsaved_values:	1✔
198	self._unsaved_values.remove(k)	1✔
199
200	# Custom unpickling method that uses `update` to update the dictionary
201	# without calling __setitem__, which would fail if any value is an empty
202	# string
203	def __setstate__(self, state: Dict[str, Any]) -> None:	1✔
204	self.update(state)	1✔
205
206	# Custom pickling method to ensure the instance is pickled as a custom
207	# class and not as a dict, otherwise __setstate__ would not be called when
208	# unpickling.
209	def __reduce__(self) -> Tuple[Any, ...]:	1✔
210	reduce_value = (	1✔
211	type(self), # callable
212	( # args
213	self.get("id", None),
214	self.api_key,
215	self.stripe_version,
216	self.stripe_account,
217	),
218	dict(self), # state
219	)
220	return reduce_value	1✔
221
222	@classmethod	1✔
223	def construct_from(	1✔
224	cls,
225	values: Dict[str, Any],
226	key: Optional[str],
227	stripe_version: Optional[str] = None,
228	stripe_account: Optional[str] = None,
229	last_response: Optional[StripeResponse] = None,
230	) -> Self:
231	instance = cls(	1✔
232	values.get("id"),
233	api_key=key,
234	stripe_version=stripe_version,
235	stripe_account=stripe_account,
236	last_response=last_response,
237	)
238	instance.refresh_from(	1✔
239	values,
240	api_key=key,
241	stripe_version=stripe_version,
242	stripe_account=stripe_account,
243	last_response=last_response,
244	)
245	return instance	1✔
246
247	def refresh_from(	1✔
248	self,
249	values: Dict[str, Any],
250	api_key: Optional[str] = None,
251	partial: Optional[bool] = False,
252	stripe_version: Optional[str] = None,
253	stripe_account: Optional[str] = None,
254	last_response: Optional[StripeResponse] = None,
255	) -> None:
256	self.api_key = api_key or getattr(values, "api_key", None)	1✔
257	self.stripe_version = stripe_version or getattr(	1✔
258	values, "stripe_version", None
259	)
260	self.stripe_account = stripe_account or getattr(	1✔
261	values, "stripe_account", None
262	)
263	self._last_response = last_response or getattr(	1✔
264	values, "_last_response", None
265	)
266
267	# Wipe old state before setting new. This is useful for e.g.
268	# updating a customer, where there is no persistent card
269	# parameter. Mark those values which don't persist as transient
270	if partial:	1✔
271	self._unsaved_values = self._unsaved_values - set(values)	1✔
272	else:
273	removed = set(self.keys()) - set(values)	1✔
274	self._transient_values = self._transient_values \| removed	1✔
275	self._unsaved_values = set()	1✔
276	self.clear()	1✔
277
278	self._transient_values = self._transient_values - set(values)	1✔
279
280	for k, v in values.items():	1✔
281	inner_class = self._get_inner_class_type(k)	1✔
282	is_dict = self._get_inner_class_is_beneath_dict(k)	1✔
283	if is_dict:	1!
284	obj = {	×
285	k: None
286	if v is None
287	else cast(
288	StripeObject,
289	_util.convert_to_stripe_object(
290	v,
291	api_key,
292	stripe_version,
293	stripe_account,
294	None,
295	inner_class,
296	),
297	)
298	for k, v in v.items()
299	}
300	else:
301	obj = cast(	1✔
302	Union[StripeObject, List[StripeObject]],
303	_util.convert_to_stripe_object(
304	v,
305	api_key,
306	stripe_version,
307	stripe_account,
308	None,
309	inner_class,
310	),
311	)
312	super(StripeObject, self).__setitem__(k, obj)	1✔
313
314	self._previous = values	1✔
315
316	@classmethod	1✔
317	@_util.deprecated(	1✔
318	"This will be removed in a future version of stripe-python."
319	)
320	def api_base(cls) -> Optional[str]:	1✔
321	return None	1✔
322
323	def request(	1✔
324	self,
325	method: Literal["get", "post", "delete"],
326	url: str,
327	params: Optional[Dict[str, Any]] = None,
328	headers: Optional[Dict[str, str]] = None,
329	) -> "StripeObject":
330	return StripeObject._request(	1✔
331	self, method, url, headers=headers, params=params
332	)
333
334	# The `method_` and `url_` arguments are suffixed with an underscore to
335	# avoid conflicting with actual request parameters in `params`.
336	def _request(	1✔
337	self,
338	method_: Literal["get", "post", "delete"],
339	url_: str,
340	api_key: Optional[str] = None,
341	idempotency_key: Optional[str] = None,
342	stripe_version: Optional[str] = None,
343	stripe_account: Optional[str] = None,
344	headers: Optional[Dict[str, str]] = None,
345	params: Optional[Mapping[str, Any]] = None,
346	) -> "StripeObject":
347	params = None if params is None else dict(params)	1✔
348	api_key = _util.read_special_variable(params, "api_key", api_key)	1✔
349	idempotency_key = _util.read_special_variable(	1✔
350	params, "idempotency_key", idempotency_key
351	)
352	stripe_version = _util.read_special_variable(	1✔
353	params, "stripe_version", stripe_version
354	)
355	stripe_account = _util.read_special_variable(	1✔
356	params, "stripe_account", stripe_account
357	)
358	headers = _util.read_special_variable(params, "headers", headers)	1✔
359
360	stripe_account = stripe_account or self.stripe_account	1✔
361	stripe_version = stripe_version or self.stripe_version	1✔
362	api_key = api_key or self.api_key	1✔
363	params = params or self._retrieve_params	1✔
364	api_base = None	1✔
365	with warnings.catch_warnings():	1✔
366	warnings.simplefilter("ignore")	1✔
367	api_base = self.api_base() # pyright: ignore[reportDeprecated]	1✔
368
369	requestor = stripe.APIRequestor(	1✔
370	key=api_key,
371	api_base=api_base,
372	api_version=stripe_version,
373	account=stripe_account,
374	)
375
376	if idempotency_key is not None:	1✔
377	headers = {} if headers is None else headers.copy()	1✔
378	headers.update(_util.populate_headers(idempotency_key))	1✔
379
380	response, api_key = requestor.request(method_, url_, params, headers)	1✔
381
382	return _util.convert_to_stripe_object(	1✔
383	response, api_key, stripe_version, stripe_account, params
384	)
385
386	def request_stream(	1✔
387	self,
388	method: str,
389	url: str,
390	params: Optional[Mapping[str, Any]] = None,
391	headers: Optional[Mapping[str, str]] = None,
392	) -> StripeStreamResponse:
393	if params is None:	1!
394	params = self._retrieve_params	×
395	api_base = None	1✔
396	with warnings.catch_warnings():	1✔
397	warnings.simplefilter("ignore")	1✔
398	api_base = self.api_base() # pyright: ignore[reportDeprecated]	1✔
399	requestor = stripe.APIRequestor(	1✔
400	key=self.api_key,
401	api_base=api_base,
402	api_version=self.stripe_version,
403	account=self.stripe_account,
404	)
405	response, _ = requestor.request_stream(method, url, params, headers)	1✔
406
407	return response	1✔
408
409	def __repr__(self) -> str:	1✔
410	ident_parts = [type(self).__name__]	1✔
411
412	obj_str = self.get("object")	1✔
413	if isinstance(obj_str, str):	1!
414	ident_parts.append(obj_str)	1✔
415
416	if isinstance(self.get("id"), str):	1!
417	ident_parts.append("id=%s" % (self.get("id"),))	1✔
418
419	unicode_repr = "<%s at %s> JSON: %s" % (	1✔
420	" ".join(ident_parts),
421	hex(id(self)),
422	str(self),
423	)
424	return unicode_repr	1✔
425
426	def __str__(self) -> str:	1✔
427	return json.dumps(	1✔
428	self._to_dict_recursive(),
429	sort_keys=True,
430	indent=2,
431	cls=self._ReprJSONEncoder,
432	)
433
434	@_util.deprecated(	1✔
435	"Deprecated. The public interface will be removed in a future version."
436	)
437	def to_dict(self) -> Dict[str, Any]:	1✔
438	return dict(self)	×
439
440	def _to_dict_recursive(self) -> Dict[str, Any]:	1✔
441	def maybe_to_dict_recursive(	1✔
442	value: Optional[Union[StripeObject, Dict[str, Any]]]
443	) -> Optional[Dict[str, Any]]:
444	if value is None:	1✔
445	return None	1✔
446	elif isinstance(value, StripeObject):	1✔
447	return value._to_dict_recursive()	1✔
448	else:
449	return value	1✔
450
451	return {	1✔
452	key: list(map(maybe_to_dict_recursive, cast(List[Any], value)))
453	if isinstance(value, list)
454	else maybe_to_dict_recursive(value)
455	for key, value in dict(self).items()
456	}
457
458	@_util.deprecated(	1✔
459	"For internal stripe-python use only. The public interface will be removed in a future version."
460	)
461	def to_dict_recursive(self) -> Dict[str, Any]:	1✔
462	return self._to_dict_recursive()	1✔
463
464	@property	1✔
465	@_util.deprecated(	1✔
466	"For internal stripe-python use only. The public interface will be removed in a future version."
467	)
468	def stripe_id(self) -> Optional[str]:	1✔
469	return getattr(self, "id")	×
470
471	def serialize(self, previous: Optional[Dict[str, Any]]) -> Dict[str, Any]:	1✔
472	params: Dict[str, Any] = {}	1✔
473	unsaved_keys = self._unsaved_values or set()	1✔
474	previous = previous or self._previous or {}	1✔
475
476	for k, v in self.items():	1✔
477	if k == "id" or k.startswith("_"):	1✔
478	continue	1✔
479	elif isinstance(v, stripe.APIResource):	1✔
480	continue	1✔
481	elif hasattr(v, "serialize"):	1✔
482	child = v.serialize(previous.get(k, None))	1✔
483	if child != {}:	1✔
484	params[k] = child	1✔
485	elif k in unsaved_keys:	1✔
486	params[k] = _compute_diff(v, previous.get(k, None))	1✔
487	elif k == "additional_owners" and v is not None:	1✔
488	params[k] = _serialize_list(v, previous.get(k, None))	1✔
489
490	return params	1✔
491
492	# This class overrides __setitem__ to throw exceptions on inputs that it
493	# doesn't like. This can cause problems when we try to copy an object
494	# wholesale because some data that's returned from the API may not be valid
495	# if it was set to be set manually. Here we override the class' copy
496	# arguments so that we can bypass these possible exceptions on __setitem__.
497	def __copy__(self) -> "StripeObject":	1✔
498	copied = StripeObject(	1✔
499	self.get("id"),
500	self.api_key,
501	stripe_version=self.stripe_version,
502	stripe_account=self.stripe_account,
503	)
504
505	copied._retrieve_params = self._retrieve_params	1✔
506
507	for k, v in self.items():	1✔
508	# Call parent's __setitem__ to avoid checks that we've added in the
509	# overridden version that can throw exceptions.
510	super(StripeObject, copied).__setitem__(k, v)	1✔
511
512	return copied	1✔
513
514	# This class overrides __setitem__ to throw exceptions on inputs that it
515	# doesn't like. This can cause problems when we try to copy an object
516	# wholesale because some data that's returned from the API may not be valid
517	# if it was set to be set manually. Here we override the class' copy
518	# arguments so that we can bypass these possible exceptions on __setitem__.
519	def __deepcopy__(self, memo: Dict[int, Any]) -> "StripeObject":	1✔
520	copied = self.__copy__()	1✔
521	memo[id(self)] = copied	1✔
522
523	for k, v in self.items():	1✔
524	# Call parent's __setitem__ to avoid checks that we've added in the
525	# overridden version that can throw exceptions.
526	super(StripeObject, copied).__setitem__(k, deepcopy(v, memo))	1✔
527
528	return copied	1✔
529
530	_inner_class_types: ClassVar[Dict[str, Type["StripeObject"]]] = {}	1✔
531	_inner_class_dicts: ClassVar[List[str]] = []	1✔
532
533	def _get_inner_class_type(	1✔
534	self, field_name: str
535	) -> Optional[Type["StripeObject"]]:
536	return self._inner_class_types.get(field_name)	1✔
537
538	def _get_inner_class_is_beneath_dict(self, field_name: str):	1✔
539	return field_name in self._inner_class_dicts	1✔

stripe / stripe-python / 7175125064

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous