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
92
93
94
95
96
97
98
99
100
101
102
103
104
|
from logging import Logger
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlencode
from requests import Response
from twilio.base.exceptions import TwilioException
from twilio.http.request import Request as TwilioRequest
from twilio.http.response import Response as TwilioResponse
class HttpClient(object):
def __init__(self, logger: Logger, is_async: bool, timeout: Optional[float] = None):
"""
Constructor for the abstract HTTP client
:param logger
:param is_async: Whether the client supports async request calls.
:param timeout: Timeout for the requests.
Timeout should never be zero (0) or less.
"""
self.logger = logger
self.is_async = is_async
if timeout is not None and timeout <= 0:
raise ValueError(timeout)
self.timeout = timeout
self._test_only_last_request: Optional[TwilioRequest] = None
self._test_only_last_response: Optional[TwilioResponse] = None
"""
An abstract class representing an HTTP client.
"""
def request(
self,
method: str,
uri: str,
params: Optional[Dict[str, object]] = None,
data: Optional[Dict[str, object]] = None,
headers: Optional[Dict[str, str]] = None,
auth: Optional[Tuple[str, str]] = None,
timeout: Optional[float] = None,
allow_redirects: bool = False,
) -> TwilioResponse:
"""
Make an HTTP request.
"""
raise TwilioException("HttpClient is an abstract class")
def log_request(self, kwargs: Dict[str, Any]) -> None:
"""
Logs the HTTP request
"""
self.logger.info("-- BEGIN Twilio API Request --")
if kwargs["params"]:
self.logger.info(
"{} Request: {}?{}".format(
kwargs["method"], kwargs["url"], urlencode(kwargs["params"])
)
)
self.logger.info("Query Params: {}".format(kwargs["params"]))
else:
self.logger.info("{} Request: {}".format(kwargs["method"], kwargs["url"]))
if kwargs["headers"]:
self.logger.info("Headers:")
for key, value in kwargs["headers"].items():
# Do not log authorization headers
if "authorization" not in key.lower():
self.logger.info("{} : {}".format(key, value))
self.logger.info("-- END Twilio API Request --")
def log_response(self, status_code: int, response: Response) -> None:
"""
Logs the HTTP response
"""
self.logger.info("Response Status Code: {}".format(status_code))
self.logger.info("Response Headers: {}".format(response.headers))
class AsyncHttpClient(HttpClient):
"""
An abstract class representing an asynchronous HTTP client.
"""
async def request(
self,
method: str,
uri: str,
params: Optional[Dict[str, object]] = None,
data: Optional[Dict[str, object]] = None,
headers: Optional[Dict[str, str]] = None,
auth: Optional[Tuple[str, str]] = None,
timeout: Optional[float] = None,
allow_redirects: bool = False,
) -> TwilioResponse:
"""
Make an asynchronous HTTP request.
"""
raise TwilioException("AsyncHttpClient is an abstract class")
|
