10615964443

Committed 29 Aug 2024 01:23PM UTC coverage: 96.379%. Remained the same

Build # 10615964443

Build Type

push

github

Committed by

web-flow

Commit Message

Merge from master after the changes to not pass api_mode from individual methods (#1384)

* Changed V1File implementation to pass content_type in options (#1382)

* No need for api_mode to be passed around from individual API methods (#1383)

* Generated code

* We have separate tests for preview mode

* Remove unused import

---------

Co-authored-by: prathmesh-stripe <165320323+prathmesh-stripe@users.noreply.github.com>

Run Details

1348 of 2043 branches covered (65.98%)

Branch coverage included in aggregate %.

11 of 12 new or added lines in 6 files covered. (91.67%)

3 existing lines in 2 files now uncovered.

84867 of 87411 relevant lines covered (97.09%)

0.97 hits per line

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

91.85

/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 import _util

from stripe._stripe_response import (
    StripeResponse,
    StripeStreamResponse,
    StripeStreamResponseAsync,
)
from stripe._encode import _encode_datetime  # pyright: ignore
from stripe._request_options import extract_options_from_dict
from stripe._api_mode import ApiMode
from stripe._base_address import BaseAddress

if TYPE_CHECKING:
    from stripe 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(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)

    _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,
    ):
        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

        self._requestor = (
            stripe._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

    # 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 in {"api_key", "stripe_account", "stripe_version"}:
                self._requestor = self._requestor._replace_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()

        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,
        *,
        api_mode: ApiMode = "V1",
    ) -> Self:
        return cls._construct_from(
            values=values,
            requestor=stripe._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: Dict[str, Any],
        last_response: Optional[StripeResponse] = None,
        requestor: "_APIRequestor",
        api_mode: ApiMode,
    ) -> Self:
        instance = cls(
            values.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._replace_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: Dict[str, Any],
        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
        )

        # 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(  # 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,
                    ),
                )
            super(StripeObject, self).__setitem__(k, obj)

        self._previous = values

    @_util.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 = 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[Mapping[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

    _field_remappings: ClassVar[Dict[str, str]] = {}

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

stripe / stripe-python / 10615964443

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