core.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381
  1. import asyncio
  2. import logging
  3. import os
  4. import ssl
  5. import time
  6. from enum import Enum
  7. from typing import Any, Dict, Optional
  8. from urllib.parse import urlencode, urlparse
  9. import aiohttp
  10. import certifi
  11. from requests import Session, PreparedRequest, adapters, status_codes
  12. from Tea.exceptions import RequiredArgumentException, RetryError
  13. from Tea.model import TeaModel
  14. from Tea.request import TeaRequest
  15. from Tea.response import TeaResponse
  16. from Tea.stream import BaseStream
  17. DEFAULT_CONNECT_TIMEOUT = 5000
  18. DEFAULT_READ_TIMEOUT = 10000
  19. DEFAULT_POOL_SIZE = 10
  20. logger = logging.getLogger('alibabacloud-tea')
  21. logger.setLevel(logging.DEBUG)
  22. ch = logging.StreamHandler()
  23. logger.addHandler(ch)
  24. class TLSVersion(Enum):
  25. TLSv1 = 'TLSv1'
  26. TLSv1_1 = 'TLSv1.1'
  27. TLSv1_2 = 'TLSv1.2'
  28. TLSv1_3 = 'TLSv1.3'
  29. class _TLSAdapter(adapters.HTTPAdapter):
  30. """A HTTPAdapter that uses an arbitrary TLS version."""
  31. def __init__(self, ssl_context=None, **kwargs):
  32. self.ssl_context = ssl_context
  33. super().__init__(**kwargs)
  34. def init_poolmanager(self, *args, **kwargs):
  35. """Override the init_poolmanager method to set the SSL."""
  36. kwargs['ssl_context'] = self.ssl_context
  37. super().init_poolmanager(*args, **kwargs)
  38. class TeaCore:
  39. _sessions = {}
  40. http_adapter = adapters.HTTPAdapter(pool_connections=DEFAULT_POOL_SIZE, pool_maxsize=DEFAULT_POOL_SIZE * 4)
  41. https_adapter = adapters.HTTPAdapter(pool_connections=DEFAULT_POOL_SIZE, pool_maxsize=DEFAULT_POOL_SIZE * 4)
  42. @staticmethod
  43. def _set_tls_minimum_version(sls_context, tls_min_version):
  44. context = sls_context
  45. if tls_min_version is not None:
  46. if tls_min_version == 'TLSv1':
  47. context.minimum_version = ssl.TLSVersion.TLSv1
  48. elif tls_min_version == 'TLSv1.1':
  49. context.minimum_version = ssl.TLSVersion.TLSv1_1
  50. elif tls_min_version == 'TLSv1.2':
  51. context.minimum_version = ssl.TLSVersion.TLSv1_2
  52. elif tls_min_version == 'TLSv1.3':
  53. context.minimum_version = ssl.TLSVersion.TLSv1_3
  54. return context
  55. @staticmethod
  56. def get_adapter(prefix, tls_min_version: str = None):
  57. ca_cert = certifi.where()
  58. context = ssl.create_default_context()
  59. if ca_cert and prefix.upper() == 'HTTPS':
  60. context = TeaCore._set_tls_minimum_version(context, tls_min_version)
  61. context.load_verify_locations(ca_cert)
  62. adapter = _TLSAdapter(ssl_context=context, pool_connections=DEFAULT_POOL_SIZE,
  63. pool_maxsize=DEFAULT_POOL_SIZE * 4)
  64. return adapter
  65. @staticmethod
  66. def _get_session(session_key: str, protocol: str, tls_min_version: str = None, verify: bool = True):
  67. if session_key not in TeaCore._sessions:
  68. session = Session()
  69. adapter = TeaCore.get_adapter(protocol, tls_min_version)
  70. if protocol.upper() == 'HTTPS':
  71. if verify:
  72. session.mount('https://', adapter)
  73. else:
  74. session.mount('https://', TeaCore.https_adapter)
  75. else:
  76. session.mount('http://', adapter)
  77. TeaCore._sessions[session_key] = session
  78. return TeaCore._sessions[session_key]
  79. @staticmethod
  80. def _prepare_http_debug(request, symbol):
  81. base = ''
  82. for key, value in request.headers.items():
  83. base += f'\n{symbol} {key} : {value}'
  84. return base
  85. @staticmethod
  86. def _do_http_debug(request, response):
  87. # logger the request
  88. url = urlparse(request.url)
  89. request_base = f'\n> {request.method.upper()} {url.path + url.query} HTTP/1.1'
  90. logger.debug(request_base + TeaCore._prepare_http_debug(request, '>'))
  91. # logger the response
  92. response_base = f'\n< HTTP/1.1 {response.status_code}' \
  93. f' {status_codes._codes.get(response.status_code)[0].upper()}'
  94. logger.debug(response_base + TeaCore._prepare_http_debug(response, '<'))
  95. @staticmethod
  96. def compose_url(request):
  97. host = request.headers.get('host')
  98. if not host:
  99. raise RequiredArgumentException('endpoint')
  100. else:
  101. host = host.rstrip('/')
  102. protocol = f'{request.protocol.lower()}://'
  103. pathname = request.pathname
  104. if host.startswith(('http://', 'https://')):
  105. protocol = ''
  106. if request.port == 80:
  107. port = ''
  108. else:
  109. port = f':{request.port}'
  110. url = protocol + host + port + pathname
  111. if request.query:
  112. if "?" in url:
  113. if not url.endswith("&"):
  114. url += "&"
  115. else:
  116. url += "?"
  117. encode_query = {}
  118. for key in request.query:
  119. value = request.query[key]
  120. if value is not None:
  121. encode_query[key] = str(value)
  122. url += urlencode(encode_query)
  123. return url.rstrip("?&")
  124. @staticmethod
  125. async def async_do_action(
  126. request: TeaRequest,
  127. runtime_option=None
  128. ) -> TeaResponse:
  129. runtime_option = runtime_option or {}
  130. url = TeaCore.compose_url(request)
  131. verify = not runtime_option.get('ignoreSSL', False)
  132. tls_min_version = runtime_option.get('tlsMinVersion')
  133. if isinstance(tls_min_version, Enum):
  134. tls_min_version = tls_min_version.value
  135. timeout = runtime_option.get('timeout')
  136. connect_timeout = runtime_option.get('connectTimeout') or timeout or DEFAULT_CONNECT_TIMEOUT
  137. read_timeout = runtime_option.get('readTimeout') or timeout or DEFAULT_READ_TIMEOUT
  138. connect_timeout, read_timeout = (int(connect_timeout) / 1000, int(read_timeout) / 1000)
  139. proxy = None
  140. if request.protocol.upper() == 'HTTP':
  141. proxy = runtime_option.get('httpProxy')
  142. if not proxy:
  143. proxy = os.environ.get('HTTP_PROXY') or os.environ.get('http_proxy')
  144. elif request.protocol.upper() == 'HTTPS':
  145. proxy = runtime_option.get('httpsProxy')
  146. if not proxy:
  147. proxy = os.environ.get('HTTPS_PROXY') or os.environ.get('https_proxy')
  148. connector = None
  149. ca_cert = certifi.where()
  150. if ca_cert and request.protocol.upper() == 'HTTPS':
  151. ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
  152. ssl_context = TeaCore._set_tls_minimum_version(ssl_context, tls_min_version)
  153. ssl_context.load_verify_locations(ca_cert)
  154. connector = aiohttp.TCPConnector(
  155. ssl=ssl_context,
  156. )
  157. else:
  158. verify = False
  159. timeout = aiohttp.ClientTimeout(
  160. sock_read=read_timeout,
  161. sock_connect=connect_timeout
  162. )
  163. async with aiohttp.ClientSession(
  164. connector=connector
  165. ) as s:
  166. body = b''
  167. if isinstance(request.body, BaseStream):
  168. for content in request.body:
  169. body += content
  170. elif isinstance(request.body, str):
  171. body = request.body.encode('utf-8')
  172. else:
  173. body = request.body
  174. try:
  175. async with s.request(request.method, url,
  176. data=body,
  177. headers=request.headers,
  178. ssl=verify,
  179. proxy=proxy,
  180. timeout=timeout) as response:
  181. tea_resp = TeaResponse()
  182. tea_resp.body = await response.read()
  183. tea_resp.headers = {k.lower(): v for k, v in response.headers.items()}
  184. tea_resp.status_code = response.status
  185. tea_resp.status_message = response.reason
  186. tea_resp.response = response
  187. except IOError as e:
  188. raise RetryError(str(e))
  189. return tea_resp
  190. @staticmethod
  191. def do_action(
  192. request: TeaRequest,
  193. runtime_option=None
  194. ) -> TeaResponse:
  195. url = TeaCore.compose_url(request)
  196. runtime_option = runtime_option or {}
  197. verify = not runtime_option.get('ignoreSSL', False)
  198. tls_min_version = runtime_option.get('tlsMinVersion')
  199. if isinstance(tls_min_version, Enum):
  200. tls_min_version = tls_min_version.value
  201. if verify:
  202. verify = runtime_option.get('ca', True) if runtime_option.get('ca', True) is not None else True
  203. cert = runtime_option.get('cert', None)
  204. timeout = runtime_option.get('timeout')
  205. connect_timeout = runtime_option.get('connectTimeout') or timeout or DEFAULT_CONNECT_TIMEOUT
  206. read_timeout = runtime_option.get('readTimeout') or timeout or DEFAULT_READ_TIMEOUT
  207. timeout = (int(connect_timeout) / 1000, int(read_timeout) / 1000)
  208. if isinstance(request.body, str):
  209. request.body = request.body.encode('utf-8')
  210. p = PreparedRequest()
  211. p.prepare(
  212. method=request.method.upper(),
  213. url=url,
  214. data=request.body,
  215. headers=request.headers,
  216. )
  217. proxies = {}
  218. http_proxy = runtime_option.get('httpProxy')
  219. https_proxy = runtime_option.get('httpsProxy')
  220. no_proxy = runtime_option.get('noProxy')
  221. if not http_proxy:
  222. http_proxy = os.environ.get('HTTP_PROXY') or os.environ.get('http_proxy')
  223. if not https_proxy:
  224. https_proxy = os.environ.get('HTTPS_PROXY') or os.environ.get('https_proxy')
  225. if http_proxy:
  226. proxies['http'] = http_proxy
  227. if https_proxy:
  228. proxies['https'] = https_proxy
  229. if no_proxy:
  230. proxies['no_proxy'] = no_proxy
  231. host = request.headers.get('host')
  232. host = host.rstrip('/')
  233. session_key = f'{request.protocol.lower()}://{host}:{request.port}'
  234. session = TeaCore._get_session(session_key=session_key, protocol=request.protocol,
  235. tls_min_version=tls_min_version, verify=verify)
  236. try:
  237. resp = session.send(
  238. p,
  239. proxies=proxies,
  240. timeout=timeout,
  241. verify=verify,
  242. cert=cert,
  243. )
  244. except IOError as e:
  245. raise RetryError(str(e))
  246. debug = runtime_option.get('debug') or os.getenv('DEBUG')
  247. if debug and debug.lower() == 'sdk':
  248. TeaCore._do_http_debug(p, resp)
  249. response = TeaResponse()
  250. response.status_message = resp.reason
  251. response.status_code = resp.status_code
  252. response.headers = {k.lower(): v for k, v in resp.headers.items()}
  253. response.body = resp.content
  254. response.response = resp
  255. return response
  256. @staticmethod
  257. def get_response_body(resp) -> str:
  258. return resp.content.decode("utf-8")
  259. @staticmethod
  260. def allow_retry(dic, retry_times, now=None) -> bool:
  261. if retry_times == 0:
  262. return True
  263. if dic is None or not dic.__contains__("maxAttempts") or \
  264. dic.get('retryable') is not True and retry_times >= 1:
  265. return False
  266. else:
  267. retry = 0 if dic.get("maxAttempts") is None else int(
  268. dic.get("maxAttempts"))
  269. return retry >= retry_times
  270. @staticmethod
  271. def get_backoff_time(dic, retry_times) -> int:
  272. default_back_off_time = 0
  273. if dic is None or not dic.get("policy") or dic.get("policy") == "no":
  274. return default_back_off_time
  275. back_off_time = dic.get('period', default_back_off_time)
  276. if not isinstance(back_off_time, int) and \
  277. not (isinstance(back_off_time, str) and back_off_time.isdigit()):
  278. return default_back_off_time
  279. back_off_time = int(back_off_time)
  280. if back_off_time < 0:
  281. return retry_times
  282. return back_off_time
  283. @staticmethod
  284. async def sleep_async(t):
  285. await asyncio.sleep(t)
  286. @staticmethod
  287. def sleep(t):
  288. time.sleep(t)
  289. @staticmethod
  290. def is_retryable(ex) -> bool:
  291. return isinstance(ex, RetryError)
  292. @staticmethod
  293. def bytes_readable(body):
  294. return body
  295. @staticmethod
  296. def merge(*dic_list) -> dict:
  297. dic_result = {}
  298. for item in dic_list:
  299. if isinstance(item, dict):
  300. dic_result.update(item)
  301. elif isinstance(item, TeaModel):
  302. dic_result.update(item.to_map())
  303. return dic_result
  304. @staticmethod
  305. def to_map(model: Optional[TeaModel]) -> Dict[str, Any]:
  306. if isinstance(model, TeaModel):
  307. return model.to_map()
  308. else:
  309. return dict()
  310. @staticmethod
  311. def from_map(
  312. model: TeaModel,
  313. dic: Dict[str, Any]
  314. ) -> TeaModel:
  315. if isinstance(model, TeaModel):
  316. try:
  317. return model.from_map(dic)
  318. except Exception:
  319. model._map = dic
  320. return model
  321. else:
  322. return model