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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import os
import logging
from typing import Dict, Optional, Tuple
from requests import Request, Session, hooks
from requests.adapters import HTTPAdapter
from twilio.http import HttpClient
from twilio.http.request import Request as TwilioRequest
from twilio.http.response import Response
_logger = logging.getLogger("twilio.http_client")
class TwilioHttpClient(HttpClient):
"""
General purpose HTTP Client for interacting with the Twilio API
"""
def __init__(
self,
pool_connections: bool = True,
request_hooks: Optional[Dict[str, object]] = None,
timeout: Optional[float] = None,
logger: logging.Logger = _logger,
proxy: Optional[Dict[str, str]] = None,
max_retries: Optional[int] = None,
):
"""
Constructor for the TwilioHttpClient
:param pool_connections
:param request_hooks
:param timeout: Timeout for the requests.
Timeout should never be zero (0) or less
:param logger
:param proxy: Http proxy for the requests session
:param max_retries: Maximum number of retries each request should attempt
"""
super().__init__(logger, False, timeout)
self.session = Session() if pool_connections else None
if self.session and max_retries is not None:
self.session.mount("https://", HTTPAdapter(max_retries=max_retries))
if self.session is not None:
self.session.mount(
"https://", HTTPAdapter(pool_maxsize=min(32, os.cpu_count() + 4))
)
self.request_hooks = request_hooks or hooks.default_hooks()
self.proxy = proxy if proxy else {}
def request(
self,
method: str,
url: 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,
) -> Response:
"""
Make an HTTP Request with parameters provided.
:param method: The HTTP method to use
:param url: The URL to request
:param params: Query parameters to append to the URL
:param data: Parameters to go in the body of the HTTP request
:param headers: HTTP Headers to send with the request
:param auth: Basic Auth arguments
:param timeout: Socket/Read timeout for the request
:param allow_redirects: Whether to allow redirects
See the requests documentation for explanation of all these parameters
:return: An HTTP response
"""
if timeout is None:
timeout = self.timeout
elif timeout <= 0:
raise ValueError(timeout)
kwargs = {
"method": method.upper(),
"url": url,
"params": params,
"headers": headers,
"auth": auth,
"hooks": self.request_hooks,
}
if headers and headers.get("Content-Type") == "application/json":
kwargs["json"] = data
elif headers and headers.get("Content-Type") == "application/scim+json":
kwargs["json"] = data
else:
kwargs["data"] = data
self.log_request(kwargs)
self._test_only_last_response = None
session = self.session or Session()
request = Request(**kwargs)
self._test_only_last_request = TwilioRequest(**kwargs)
prepped_request = session.prepare_request(request)
settings = session.merge_environment_settings(
prepped_request.url, self.proxy, None, None, None
)
response = session.send(
prepped_request,
allow_redirects=allow_redirects,
timeout=timeout,
**settings,
)
self.log_response(response.status_code, response)
self._test_only_last_response = Response(
int(response.status_code), response.text, response.headers
)
return self._test_only_last_response
|
