11027499654

Committed 25 Sep 2024 06:24AM UTC coverage: 96.229% (-0.09%) from 96.317%

Build # 11027499654

Build Type

Pull #1397

github

Committed by

web-flow

Commit Message

Merge 1919d72b0 into 3587026b8

Pull Request Pull Request #1397: Next major infra

Run Details

1333 of 1969 branches covered (67.7%)

Branch coverage included in aggregate %.

512 of 587 new or added lines in 40 files covered. (87.22%)

1 existing line in 1 file now uncovered.

67650 of 69717 relevant lines covered (97.04%)

0.97 hits per line

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

88.75

/stripe/_api_resource.py

from typing_extensions import Literal, Self

from stripe._error import InvalidRequestError
from stripe._stripe_object import StripeObject
from stripe._request_options import extract_options_from_dict
from stripe._api_mode import ApiMode
from stripe._base_address import BaseAddress
from stripe._api_requestor import _APIRequestor
from stripe import _util
from urllib.parse import quote_plus
from typing import (
    Any,
    ClassVar,
    Generic,
    List,
    Optional,
    TypeVar,
    cast,
    Mapping,
)

T = TypeVar("T", bound=StripeObject)


class APIResource(StripeObject, Generic[T]):
    OBJECT_NAME: ClassVar[str]

    @classmethod
    @_util.deprecated(
        "This method is deprecated and will be removed in a future version of stripe-python. Child classes of APIResource should define their own `retrieve` and use APIResource._request directly."
    )
    def retrieve(cls, id, **params) -> T:
        instance = cls(id, **params)
        instance.refresh()
        return cast(T, instance)

    def refresh(self) -> Self:
        return self._request_and_refresh("get", self.instance_url())

    async def refresh_async(self) -> Self:
        return await self._request_and_refresh_async(
            "get", self.instance_url()
        )

    @classmethod
    def class_url(cls) -> str:
        if cls == APIResource:
            raise NotImplementedError(
                "APIResource is an abstract class.  You should perform "
                "actions on its subclasses (e.g. Charge, Customer)"
            )
        # Namespaces are separated in object names with periods (.) and in URLs
        # with forward slashes (/), so replace the former with the latter.
        base = cls.OBJECT_NAME.replace(".", "/")
        return "/v1/%ss" % (base,)

    def instance_url(self) -> str:
        id = self.get("id")

        if not isinstance(id, str):
            raise InvalidRequestError(
                "Could not determine which URL to request: %s instance "
                "has invalid ID: %r, %s. ID should be of type `str` (or"
                " `unicode`)" % (type(self).__name__, id, type(id)),
                "id",
            )

        base = self.class_url()
        extn = quote_plus(id)
        return "%s/%s" % (base, extn)

    def _request(
        self,
        method,
        url,
        params=None,
        *,
        base_address: BaseAddress = "api",
        api_mode: ApiMode = "V1",
    ) -> StripeObject:
        obj = StripeObject._request(
            self,
            method,
            url,
            params=params,
            base_address=base_address,
        )

        if type(self) is type(obj):
            self._refresh_from(values=obj, api_mode=api_mode)
            return self
        else:
            return obj

    async def _request_async(
        self,
        method,
        url,
        params=None,
        *,
        base_address: BaseAddress = "api",
        api_mode: ApiMode = "V1",
    ) -> StripeObject:
        obj = await StripeObject._request_async(
            self,
            method,
            url,
            params=params,
            base_address=base_address,
        )

        if type(self) is type(obj):
            self._refresh_from(values=obj, api_mode=api_mode)
            return self
        else:
            return obj

    def _request_and_refresh(
        self,
        method: Literal["get", "post", "delete"],
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        usage: Optional[List[str]] = None,
        *,
        base_address: BaseAddress = "api",
        api_mode: ApiMode = "V1",
    ) -> Self:
        obj = StripeObject._request(
            self,
            method,
            url,
            params=params,
            base_address=base_address,
            usage=usage,
        )

        self._refresh_from(values=obj, api_mode=api_mode)
        return self

    async def _request_and_refresh_async(
        self,
        method: Literal["get", "post", "delete"],
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        usage: Optional[List[str]] = None,
        *,
        base_address: BaseAddress = "api",
        api_mode: ApiMode = "V1",
    ) -> Self:
        obj = await StripeObject._request_async(
            self,
            method,
            url,
            params=params,
            base_address=base_address,
            usage=usage,
        )

        self._refresh_from(values=obj, api_mode=api_mode)
        return self

    @classmethod
    def _static_request(
        cls,
        method_,
        url_,
        params: Optional[Mapping[str, Any]] = None,
        *,
        base_address: BaseAddress = "api",
    ):
        request_options, request_params = extract_options_from_dict(params)
        return _APIRequestor._global_instance().request(
            method_,
            url_,
            params=request_params,
            options=request_options,
            base_address=base_address,
        )

    @classmethod
    async def _static_request_async(
        cls,
        method_,
        url_,
        params: Optional[Mapping[str, Any]] = None,
        *,
        base_address: BaseAddress = "api",
    ):
        request_options, request_params = extract_options_from_dict(params)
        return await _APIRequestor._global_instance().request_async(
            method_,
            url_,
            params=request_params,
            options=request_options,
            base_address=base_address,
        )

    @classmethod
    def _static_request_stream(
        cls,
        method,
        url,
        params: Optional[Mapping[str, Any]] = None,
        *,
        base_address: BaseAddress = "api",
    ):
        request_options, request_params = extract_options_from_dict(params)
        return _APIRequestor._global_instance().request_stream(
            method,
            url,
            params=request_params,
            options=request_options,
            base_address=base_address,
        )

    @classmethod
    async def _static_request_stream_async(
        cls,
        method,
        url,
        params: Optional[Mapping[str, Any]] = None,
        *,
        base_address: BaseAddress = "api",
    ):
        request_options, request_params = extract_options_from_dict(params)
        return await _APIRequestor._global_instance().request_stream_async(
            method,
            url,
            params=request_params,
            options=request_options,
            base_address=base_address,
        )

1	from typing_extensions import Literal, Self	1✔
2
3	from stripe._error import InvalidRequestError	1✔
4	from stripe._stripe_object import StripeObject	1✔
5	from stripe._request_options import extract_options_from_dict	1✔
6	from stripe._api_mode import ApiMode	1✔
7	from stripe._base_address import BaseAddress	1✔
8	from stripe._api_requestor import _APIRequestor	1✔
9	from stripe import _util	1✔
10	from urllib.parse import quote_plus	1✔
11	from typing import (	1✔
12	Any,
13	ClassVar,
14	Generic,
15	List,
16	Optional,
17	TypeVar,
18	cast,
19	Mapping,
20	)
21
22	T = TypeVar("T", bound=StripeObject)	1✔
23
24
25	class APIResource(StripeObject, Generic[T]):	1✔
26	OBJECT_NAME: ClassVar[str]	1✔
27
28	@classmethod	1✔
29	@_util.deprecated(	1✔
30	"This method is deprecated and will be removed in a future version of stripe-python. Child classes of APIResource should define their own `retrieve` and use APIResource._request directly."
31	)
32	def retrieve(cls, id, **params) -> T:	1✔
33	instance = cls(id, **params)	1✔
34	instance.refresh()	1✔
35	return cast(T, instance)	1✔
36
37	def refresh(self) -> Self:	1✔
38	return self._request_and_refresh("get", self.instance_url())	1✔
39
40	async def refresh_async(self) -> Self:	1✔
41	return await self._request_and_refresh_async(	1✔
42	"get", self.instance_url()
43	)
44
45	@classmethod	1✔
46	def class_url(cls) -> str:	1✔
47	if cls == APIResource:	1!
48	raise NotImplementedError(	×
49	"APIResource is an abstract class. You should perform "
50	"actions on its subclasses (e.g. Charge, Customer)"
51	)
52	# Namespaces are separated in object names with periods (.) and in URLs
53	# with forward slashes (/), so replace the former with the latter.
54	base = cls.OBJECT_NAME.replace(".", "/")	1✔
55	return "/v1/%ss" % (base,)	1✔
56
57	def instance_url(self) -> str:	1✔
58	id = self.get("id")	1✔
59
60	if not isinstance(id, str):	1✔
61	raise InvalidRequestError(	1✔
62	"Could not determine which URL to request: %s instance "
63	"has invalid ID: %r, %s. ID should be of type `str` (or"
64	" `unicode`)" % (type(self).__name__, id, type(id)),
65	"id",
66	)
67
68	base = self.class_url()	1✔
69	extn = quote_plus(id)	1✔
70	return "%s/%s" % (base, extn)	1✔
71
72	def _request(	1✔
73	self,
74	method,
75	url,
76	params=None,
77	*,
78	base_address: BaseAddress = "api",
79	api_mode: ApiMode = "V1",
80	) -> StripeObject:
81	obj = StripeObject._request(	1✔
82	self,
83	method,
84	url,
85	params=params,
86	base_address=base_address,
87	)
88
89	if type(self) is type(obj):	1✔
90	self._refresh_from(values=obj, api_mode=api_mode)	1✔
91	return self	1✔
92	else:
93	return obj	1✔
94
95	async def _request_async(	1✔
96	self,
97	method,
98	url,
99	params=None,
100	*,
101	base_address: BaseAddress = "api",
102	api_mode: ApiMode = "V1",
103	) -> StripeObject:
104	obj = await StripeObject._request_async(	×
105	self,
106	method,
107	url,
108	params=params,
109	base_address=base_address,
110	)
111
112	if type(self) is type(obj):	×
NEW 113	self._refresh_from(values=obj, api_mode=api_mode)	×
114	return self	×
115	else:
116	return obj	×
117
118	def _request_and_refresh(	1✔
119	self,
120	method: Literal["get", "post", "delete"],
121	url: str,
122	params: Optional[Mapping[str, Any]] = None,
123	usage: Optional[List[str]] = None,
124	*,
125	base_address: BaseAddress = "api",
126	api_mode: ApiMode = "V1",
127	) -> Self:
128	obj = StripeObject._request(	1✔
129	self,
130	method,
131	url,
132	params=params,
133	base_address=base_address,
134	usage=usage,
135	)
136
137	self._refresh_from(values=obj, api_mode=api_mode)	1✔
138	return self	1✔
139
140	async def _request_and_refresh_async(	1✔
141	self,
142	method: Literal["get", "post", "delete"],
143	url: str,
144	params: Optional[Mapping[str, Any]] = None,
145	usage: Optional[List[str]] = None,
146	*,
147	base_address: BaseAddress = "api",
148	api_mode: ApiMode = "V1",
149	) -> Self:
150	obj = await StripeObject._request_async(	1✔
151	self,
152	method,
153	url,
154	params=params,
155	base_address=base_address,
156	usage=usage,
157	)
158
159	self._refresh_from(values=obj, api_mode=api_mode)	1✔
160	return self	1✔
161
162	@classmethod	1✔
163	def _static_request(	1✔
164	cls,
165	method_,
166	url_,
167	params: Optional[Mapping[str, Any]] = None,
168	*,
169	base_address: BaseAddress = "api",
170	):
171	request_options, request_params = extract_options_from_dict(params)	1✔
172	return _APIRequestor._global_instance().request(	1✔
173	method_,
174	url_,
175	params=request_params,
176	options=request_options,
177	base_address=base_address,
178	)
179
180	@classmethod	1✔
181	async def _static_request_async(	1✔
182	cls,
183	method_,
184	url_,
185	params: Optional[Mapping[str, Any]] = None,
186	*,
187	base_address: BaseAddress = "api",
188	):
189	request_options, request_params = extract_options_from_dict(params)	1✔
190	return await _APIRequestor._global_instance().request_async(	1✔
191	method_,
192	url_,
193	params=request_params,
194	options=request_options,
195	base_address=base_address,
196	)
197
198	@classmethod	1✔
199	def _static_request_stream(	1✔
200	cls,
201	method,
202	url,
203	params: Optional[Mapping[str, Any]] = None,
204	*,
205	base_address: BaseAddress = "api",
206	):
207	request_options, request_params = extract_options_from_dict(params)	1✔
208	return _APIRequestor._global_instance().request_stream(	1✔
209	method,
210	url,
211	params=request_params,
212	options=request_options,
213	base_address=base_address,
214	)
215
216	@classmethod	1✔
217	async def _static_request_stream_async(	1✔
218	cls,
219	method,
220	url,
221	params: Optional[Mapping[str, Any]] = None,
222	*,
223	base_address: BaseAddress = "api",
224	):
225	request_options, request_params = extract_options_from_dict(params)	1✔
226	return await _APIRequestor._global_instance().request_stream_async(	1✔
227	method,
228	url,
229	params=request_params,
230	options=request_options,
231	base_address=base_address,
232	)

stripe / stripe-python / 11027499654

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