File: _stripe_service.py

package info (click to toggle)
python-stripe 12.0.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 12,864 kB
  • sloc: python: 157,573; makefile: 13; sh: 9
file content (91 lines) | stat: -rw-r--r-- 2,358 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from stripe._api_requestor import (
    _APIRequestor,
)
from stripe._stripe_response import (
    StripeStreamResponse,
    StripeStreamResponseAsync,
)
from stripe._stripe_object import StripeObject
from stripe._request_options import RequestOptions
from stripe._base_address import BaseAddress

from typing import Any, Mapping, Optional


class StripeService(object):
    _requestor: _APIRequestor

    def __init__(self, requestor):
        self._requestor = requestor

    def _request(
        self,
        method: str,
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        options: Optional[RequestOptions] = None,
        *,
        base_address: BaseAddress,
    ) -> StripeObject:
        return self._requestor.request(
            method,
            url,
            params,
            options,
            base_address=base_address,
            usage=["stripe_client"],
        )

    async def _request_async(
        self,
        method: str,
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        options: Optional[RequestOptions] = None,
        *,
        base_address: BaseAddress,
    ) -> StripeObject:
        return await self._requestor.request_async(
            method,
            url,
            params,
            options,
            base_address=base_address,
            usage=["stripe_client"],
        )

    def _request_stream(
        self,
        method: str,
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        options: Optional[RequestOptions] = None,
        *,
        base_address: BaseAddress,
    ) -> StripeStreamResponse:
        return self._requestor.request_stream(
            method,
            url,
            params,
            options,
            base_address=base_address,
            usage=["stripe_client"],
        )

    async def _request_stream_async(
        self,
        method: str,
        url: str,
        params: Optional[Mapping[str, Any]] = None,
        options: Optional[RequestOptions] = None,
        *,
        base_address: BaseAddress,
    ) -> StripeStreamResponseAsync:
        return await self._requestor.request_stream_async(
            method,
            url,
            params,
            options,
            base_address=base_address,
            usage=["stripe_client"],
        )