client_reqrep.py 36 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127
  1. import asyncio
  2. import codecs
  3. import functools
  4. import io
  5. import re
  6. import sys
  7. import traceback
  8. import warnings
  9. from hashlib import md5, sha1, sha256
  10. from http.cookies import CookieError, Morsel, SimpleCookie
  11. from types import MappingProxyType, TracebackType
  12. from typing import (
  13. TYPE_CHECKING,
  14. Any,
  15. Dict,
  16. Iterable,
  17. List,
  18. Mapping,
  19. Optional,
  20. Tuple,
  21. Type,
  22. Union,
  23. cast,
  24. )
  25. import attr
  26. from multidict import CIMultiDict, CIMultiDictProxy, MultiDict, MultiDictProxy
  27. from yarl import URL
  28. from . import hdrs, helpers, http, multipart, payload
  29. from .abc import AbstractStreamWriter
  30. from .client_exceptions import (
  31. ClientConnectionError,
  32. ClientOSError,
  33. ClientResponseError,
  34. ContentTypeError,
  35. InvalidURL,
  36. ServerFingerprintMismatch,
  37. )
  38. from .formdata import FormData
  39. from .helpers import (
  40. PY_36,
  41. BaseTimerContext,
  42. BasicAuth,
  43. HeadersMixin,
  44. TimerNoop,
  45. noop,
  46. reify,
  47. set_result,
  48. )
  49. from .http import SERVER_SOFTWARE, HttpVersion10, HttpVersion11, StreamWriter
  50. from .log import client_logger
  51. from .streams import StreamReader
  52. from .typedefs import (
  53. DEFAULT_JSON_DECODER,
  54. JSONDecoder,
  55. LooseCookies,
  56. LooseHeaders,
  57. RawHeaders,
  58. )
  59. try:
  60. import ssl
  61. from ssl import SSLContext
  62. except ImportError: # pragma: no cover
  63. ssl = None # type: ignore
  64. SSLContext = object # type: ignore
  65. try:
  66. import cchardet as chardet
  67. except ImportError: # pragma: no cover
  68. import chardet # type: ignore
  69. __all__ = ("ClientRequest", "ClientResponse", "RequestInfo", "Fingerprint")
  70. if TYPE_CHECKING: # pragma: no cover
  71. from .client import ClientSession
  72. from .connector import Connection
  73. from .tracing import Trace
  74. json_re = re.compile(r"^application/(?:[\w.+-]+?\+)?json")
  75. @attr.s(auto_attribs=True, frozen=True, slots=True)
  76. class ContentDisposition:
  77. type: Optional[str]
  78. parameters: "MappingProxyType[str, str]"
  79. filename: Optional[str]
  80. @attr.s(auto_attribs=True, frozen=True, slots=True)
  81. class RequestInfo:
  82. url: URL
  83. method: str
  84. headers: "CIMultiDictProxy[str]"
  85. real_url: URL = attr.ib()
  86. @real_url.default
  87. def real_url_default(self) -> URL:
  88. return self.url
  89. class Fingerprint:
  90. HASHFUNC_BY_DIGESTLEN = {
  91. 16: md5,
  92. 20: sha1,
  93. 32: sha256,
  94. }
  95. def __init__(self, fingerprint: bytes) -> None:
  96. digestlen = len(fingerprint)
  97. hashfunc = self.HASHFUNC_BY_DIGESTLEN.get(digestlen)
  98. if not hashfunc:
  99. raise ValueError("fingerprint has invalid length")
  100. elif hashfunc is md5 or hashfunc is sha1:
  101. raise ValueError(
  102. "md5 and sha1 are insecure and " "not supported. Use sha256."
  103. )
  104. self._hashfunc = hashfunc
  105. self._fingerprint = fingerprint
  106. @property
  107. def fingerprint(self) -> bytes:
  108. return self._fingerprint
  109. def check(self, transport: asyncio.Transport) -> None:
  110. if not transport.get_extra_info("sslcontext"):
  111. return
  112. sslobj = transport.get_extra_info("ssl_object")
  113. cert = sslobj.getpeercert(binary_form=True)
  114. got = self._hashfunc(cert).digest()
  115. if got != self._fingerprint:
  116. host, port, *_ = transport.get_extra_info("peername")
  117. raise ServerFingerprintMismatch(self._fingerprint, got, host, port)
  118. if ssl is not None:
  119. SSL_ALLOWED_TYPES = (ssl.SSLContext, bool, Fingerprint, type(None))
  120. else: # pragma: no cover
  121. SSL_ALLOWED_TYPES = type(None)
  122. def _merge_ssl_params(
  123. ssl: Union["SSLContext", bool, Fingerprint, None],
  124. verify_ssl: Optional[bool],
  125. ssl_context: Optional["SSLContext"],
  126. fingerprint: Optional[bytes],
  127. ) -> Union["SSLContext", bool, Fingerprint, None]:
  128. if verify_ssl is not None and not verify_ssl:
  129. warnings.warn(
  130. "verify_ssl is deprecated, use ssl=False instead",
  131. DeprecationWarning,
  132. stacklevel=3,
  133. )
  134. if ssl is not None:
  135. raise ValueError(
  136. "verify_ssl, ssl_context, fingerprint and ssl "
  137. "parameters are mutually exclusive"
  138. )
  139. else:
  140. ssl = False
  141. if ssl_context is not None:
  142. warnings.warn(
  143. "ssl_context is deprecated, use ssl=context instead",
  144. DeprecationWarning,
  145. stacklevel=3,
  146. )
  147. if ssl is not None:
  148. raise ValueError(
  149. "verify_ssl, ssl_context, fingerprint and ssl "
  150. "parameters are mutually exclusive"
  151. )
  152. else:
  153. ssl = ssl_context
  154. if fingerprint is not None:
  155. warnings.warn(
  156. "fingerprint is deprecated, " "use ssl=Fingerprint(fingerprint) instead",
  157. DeprecationWarning,
  158. stacklevel=3,
  159. )
  160. if ssl is not None:
  161. raise ValueError(
  162. "verify_ssl, ssl_context, fingerprint and ssl "
  163. "parameters are mutually exclusive"
  164. )
  165. else:
  166. ssl = Fingerprint(fingerprint)
  167. if not isinstance(ssl, SSL_ALLOWED_TYPES):
  168. raise TypeError(
  169. "ssl should be SSLContext, bool, Fingerprint or None, "
  170. "got {!r} instead.".format(ssl)
  171. )
  172. return ssl
  173. @attr.s(auto_attribs=True, slots=True, frozen=True)
  174. class ConnectionKey:
  175. # the key should contain an information about used proxy / TLS
  176. # to prevent reusing wrong connections from a pool
  177. host: str
  178. port: Optional[int]
  179. is_ssl: bool
  180. ssl: Union[SSLContext, None, bool, Fingerprint]
  181. proxy: Optional[URL]
  182. proxy_auth: Optional[BasicAuth]
  183. proxy_headers_hash: Optional[int] # hash(CIMultiDict)
  184. def _is_expected_content_type(
  185. response_content_type: str, expected_content_type: str
  186. ) -> bool:
  187. if expected_content_type == "application/json":
  188. return json_re.match(response_content_type) is not None
  189. return expected_content_type in response_content_type
  190. class ClientRequest:
  191. GET_METHODS = {
  192. hdrs.METH_GET,
  193. hdrs.METH_HEAD,
  194. hdrs.METH_OPTIONS,
  195. hdrs.METH_TRACE,
  196. }
  197. POST_METHODS = {hdrs.METH_PATCH, hdrs.METH_POST, hdrs.METH_PUT}
  198. ALL_METHODS = GET_METHODS.union(POST_METHODS).union({hdrs.METH_DELETE})
  199. DEFAULT_HEADERS = {
  200. hdrs.ACCEPT: "*/*",
  201. hdrs.ACCEPT_ENCODING: "gzip, deflate",
  202. }
  203. body = b""
  204. auth = None
  205. response = None
  206. _writer = None # async task for streaming data
  207. _continue = None # waiter future for '100 Continue' response
  208. # N.B.
  209. # Adding __del__ method with self._writer closing doesn't make sense
  210. # because _writer is instance method, thus it keeps a reference to self.
  211. # Until writer has finished finalizer will not be called.
  212. def __init__(
  213. self,
  214. method: str,
  215. url: URL,
  216. *,
  217. params: Optional[Mapping[str, str]] = None,
  218. headers: Optional[LooseHeaders] = None,
  219. skip_auto_headers: Iterable[str] = frozenset(),
  220. data: Any = None,
  221. cookies: Optional[LooseCookies] = None,
  222. auth: Optional[BasicAuth] = None,
  223. version: http.HttpVersion = http.HttpVersion11,
  224. compress: Optional[str] = None,
  225. chunked: Optional[bool] = None,
  226. expect100: bool = False,
  227. loop: Optional[asyncio.AbstractEventLoop] = None,
  228. response_class: Optional[Type["ClientResponse"]] = None,
  229. proxy: Optional[URL] = None,
  230. proxy_auth: Optional[BasicAuth] = None,
  231. timer: Optional[BaseTimerContext] = None,
  232. session: Optional["ClientSession"] = None,
  233. ssl: Union[SSLContext, bool, Fingerprint, None] = None,
  234. proxy_headers: Optional[LooseHeaders] = None,
  235. traces: Optional[List["Trace"]] = None,
  236. ):
  237. if loop is None:
  238. loop = asyncio.get_event_loop()
  239. assert isinstance(url, URL), url
  240. assert isinstance(proxy, (URL, type(None))), proxy
  241. # FIXME: session is None in tests only, need to fix tests
  242. # assert session is not None
  243. self._session = cast("ClientSession", session)
  244. if params:
  245. q = MultiDict(url.query)
  246. url2 = url.with_query(params)
  247. q.extend(url2.query)
  248. url = url.with_query(q)
  249. self.original_url = url
  250. self.url = url.with_fragment(None)
  251. self.method = method.upper()
  252. self.chunked = chunked
  253. self.compress = compress
  254. self.loop = loop
  255. self.length = None
  256. if response_class is None:
  257. real_response_class = ClientResponse
  258. else:
  259. real_response_class = response_class
  260. self.response_class = real_response_class # type: Type[ClientResponse]
  261. self._timer = timer if timer is not None else TimerNoop()
  262. self._ssl = ssl
  263. if loop.get_debug():
  264. self._source_traceback = traceback.extract_stack(sys._getframe(1))
  265. self.update_version(version)
  266. self.update_host(url)
  267. self.update_headers(headers)
  268. self.update_auto_headers(skip_auto_headers)
  269. self.update_cookies(cookies)
  270. self.update_content_encoding(data)
  271. self.update_auth(auth)
  272. self.update_proxy(proxy, proxy_auth, proxy_headers)
  273. self.update_body_from_data(data)
  274. if data or self.method not in self.GET_METHODS:
  275. self.update_transfer_encoding()
  276. self.update_expect_continue(expect100)
  277. if traces is None:
  278. traces = []
  279. self._traces = traces
  280. def is_ssl(self) -> bool:
  281. return self.url.scheme in ("https", "wss")
  282. @property
  283. def ssl(self) -> Union["SSLContext", None, bool, Fingerprint]:
  284. return self._ssl
  285. @property
  286. def connection_key(self) -> ConnectionKey:
  287. proxy_headers = self.proxy_headers
  288. if proxy_headers:
  289. h = hash(
  290. tuple((k, v) for k, v in proxy_headers.items())
  291. ) # type: Optional[int]
  292. else:
  293. h = None
  294. return ConnectionKey(
  295. self.host,
  296. self.port,
  297. self.is_ssl(),
  298. self.ssl,
  299. self.proxy,
  300. self.proxy_auth,
  301. h,
  302. )
  303. @property
  304. def host(self) -> str:
  305. ret = self.url.raw_host
  306. assert ret is not None
  307. return ret
  308. @property
  309. def port(self) -> Optional[int]:
  310. return self.url.port
  311. @property
  312. def request_info(self) -> RequestInfo:
  313. headers = CIMultiDictProxy(self.headers) # type: CIMultiDictProxy[str]
  314. return RequestInfo(self.url, self.method, headers, self.original_url)
  315. def update_host(self, url: URL) -> None:
  316. """Update destination host, port and connection type (ssl)."""
  317. # get host/port
  318. if not url.raw_host:
  319. raise InvalidURL(url)
  320. # basic auth info
  321. username, password = url.user, url.password
  322. if username:
  323. self.auth = helpers.BasicAuth(username, password or "")
  324. def update_version(self, version: Union[http.HttpVersion, str]) -> None:
  325. """Convert request version to two elements tuple.
  326. parser HTTP version '1.1' => (1, 1)
  327. """
  328. if isinstance(version, str):
  329. v = [part.strip() for part in version.split(".", 1)]
  330. try:
  331. version = http.HttpVersion(int(v[0]), int(v[1]))
  332. except ValueError:
  333. raise ValueError(
  334. f"Can not parse http version number: {version}"
  335. ) from None
  336. self.version = version
  337. def update_headers(self, headers: Optional[LooseHeaders]) -> None:
  338. """Update request headers."""
  339. self.headers = CIMultiDict() # type: CIMultiDict[str]
  340. # add host
  341. netloc = cast(str, self.url.raw_host)
  342. if helpers.is_ipv6_address(netloc):
  343. netloc = f"[{netloc}]"
  344. if self.url.port is not None and not self.url.is_default_port():
  345. netloc += ":" + str(self.url.port)
  346. self.headers[hdrs.HOST] = netloc
  347. if headers:
  348. if isinstance(headers, (dict, MultiDictProxy, MultiDict)):
  349. headers = headers.items() # type: ignore
  350. for key, value in headers: # type: ignore
  351. # A special case for Host header
  352. if key.lower() == "host":
  353. self.headers[key] = value
  354. else:
  355. self.headers.add(key, value)
  356. def update_auto_headers(self, skip_auto_headers: Iterable[str]) -> None:
  357. self.skip_auto_headers = CIMultiDict(
  358. (hdr, None) for hdr in sorted(skip_auto_headers)
  359. )
  360. used_headers = self.headers.copy()
  361. used_headers.extend(self.skip_auto_headers) # type: ignore
  362. for hdr, val in self.DEFAULT_HEADERS.items():
  363. if hdr not in used_headers:
  364. self.headers.add(hdr, val)
  365. if hdrs.USER_AGENT not in used_headers:
  366. self.headers[hdrs.USER_AGENT] = SERVER_SOFTWARE
  367. def update_cookies(self, cookies: Optional[LooseCookies]) -> None:
  368. """Update request cookies header."""
  369. if not cookies:
  370. return
  371. c = SimpleCookie() # type: SimpleCookie[str]
  372. if hdrs.COOKIE in self.headers:
  373. c.load(self.headers.get(hdrs.COOKIE, ""))
  374. del self.headers[hdrs.COOKIE]
  375. if isinstance(cookies, Mapping):
  376. iter_cookies = cookies.items()
  377. else:
  378. iter_cookies = cookies # type: ignore
  379. for name, value in iter_cookies:
  380. if isinstance(value, Morsel):
  381. # Preserve coded_value
  382. mrsl_val = value.get(value.key, Morsel())
  383. mrsl_val.set(value.key, value.value, value.coded_value)
  384. c[name] = mrsl_val
  385. else:
  386. c[name] = value # type: ignore
  387. self.headers[hdrs.COOKIE] = c.output(header="", sep=";").strip()
  388. def update_content_encoding(self, data: Any) -> None:
  389. """Set request content encoding."""
  390. if not data:
  391. return
  392. enc = self.headers.get(hdrs.CONTENT_ENCODING, "").lower()
  393. if enc:
  394. if self.compress:
  395. raise ValueError(
  396. "compress can not be set " "if Content-Encoding header is set"
  397. )
  398. elif self.compress:
  399. if not isinstance(self.compress, str):
  400. self.compress = "deflate"
  401. self.headers[hdrs.CONTENT_ENCODING] = self.compress
  402. self.chunked = True # enable chunked, no need to deal with length
  403. def update_transfer_encoding(self) -> None:
  404. """Analyze transfer-encoding header."""
  405. te = self.headers.get(hdrs.TRANSFER_ENCODING, "").lower()
  406. if "chunked" in te:
  407. if self.chunked:
  408. raise ValueError(
  409. "chunked can not be set "
  410. 'if "Transfer-Encoding: chunked" header is set'
  411. )
  412. elif self.chunked:
  413. if hdrs.CONTENT_LENGTH in self.headers:
  414. raise ValueError(
  415. "chunked can not be set " "if Content-Length header is set"
  416. )
  417. self.headers[hdrs.TRANSFER_ENCODING] = "chunked"
  418. else:
  419. if hdrs.CONTENT_LENGTH not in self.headers:
  420. self.headers[hdrs.CONTENT_LENGTH] = str(len(self.body))
  421. def update_auth(self, auth: Optional[BasicAuth]) -> None:
  422. """Set basic auth."""
  423. if auth is None:
  424. auth = self.auth
  425. if auth is None:
  426. return
  427. if not isinstance(auth, helpers.BasicAuth):
  428. raise TypeError("BasicAuth() tuple is required instead")
  429. self.headers[hdrs.AUTHORIZATION] = auth.encode()
  430. def update_body_from_data(self, body: Any) -> None:
  431. if not body:
  432. return
  433. # FormData
  434. if isinstance(body, FormData):
  435. body = body()
  436. try:
  437. body = payload.PAYLOAD_REGISTRY.get(body, disposition=None)
  438. except payload.LookupError:
  439. body = FormData(body)()
  440. self.body = body
  441. # enable chunked encoding if needed
  442. if not self.chunked:
  443. if hdrs.CONTENT_LENGTH not in self.headers:
  444. size = body.size
  445. if size is None:
  446. self.chunked = True
  447. else:
  448. if hdrs.CONTENT_LENGTH not in self.headers:
  449. self.headers[hdrs.CONTENT_LENGTH] = str(size)
  450. # copy payload headers
  451. assert body.headers
  452. for (key, value) in body.headers.items():
  453. if key in self.headers:
  454. continue
  455. if key in self.skip_auto_headers:
  456. continue
  457. self.headers[key] = value
  458. def update_expect_continue(self, expect: bool = False) -> None:
  459. if expect:
  460. self.headers[hdrs.EXPECT] = "100-continue"
  461. elif self.headers.get(hdrs.EXPECT, "").lower() == "100-continue":
  462. expect = True
  463. if expect:
  464. self._continue = self.loop.create_future()
  465. def update_proxy(
  466. self,
  467. proxy: Optional[URL],
  468. proxy_auth: Optional[BasicAuth],
  469. proxy_headers: Optional[LooseHeaders],
  470. ) -> None:
  471. if proxy and not proxy.scheme == "http":
  472. raise ValueError("Only http proxies are supported")
  473. if proxy_auth and not isinstance(proxy_auth, helpers.BasicAuth):
  474. raise ValueError("proxy_auth must be None or BasicAuth() tuple")
  475. self.proxy = proxy
  476. self.proxy_auth = proxy_auth
  477. self.proxy_headers = proxy_headers
  478. def keep_alive(self) -> bool:
  479. if self.version < HttpVersion10:
  480. # keep alive not supported at all
  481. return False
  482. if self.version == HttpVersion10:
  483. if self.headers.get(hdrs.CONNECTION) == "keep-alive":
  484. return True
  485. else: # no headers means we close for Http 1.0
  486. return False
  487. elif self.headers.get(hdrs.CONNECTION) == "close":
  488. return False
  489. return True
  490. async def write_bytes(
  491. self, writer: AbstractStreamWriter, conn: "Connection"
  492. ) -> None:
  493. """Support coroutines that yields bytes objects."""
  494. # 100 response
  495. if self._continue is not None:
  496. await writer.drain()
  497. await self._continue
  498. protocol = conn.protocol
  499. assert protocol is not None
  500. try:
  501. if isinstance(self.body, payload.Payload):
  502. await self.body.write(writer)
  503. else:
  504. if isinstance(self.body, (bytes, bytearray)):
  505. self.body = (self.body,) # type: ignore
  506. for chunk in self.body:
  507. await writer.write(chunk) # type: ignore
  508. await writer.write_eof()
  509. except OSError as exc:
  510. new_exc = ClientOSError(
  511. exc.errno, "Can not write request body for %s" % self.url
  512. )
  513. new_exc.__context__ = exc
  514. new_exc.__cause__ = exc
  515. protocol.set_exception(new_exc)
  516. except asyncio.CancelledError as exc:
  517. if not conn.closed:
  518. protocol.set_exception(exc)
  519. except Exception as exc:
  520. protocol.set_exception(exc)
  521. finally:
  522. self._writer = None
  523. async def send(self, conn: "Connection") -> "ClientResponse":
  524. # Specify request target:
  525. # - CONNECT request must send authority form URI
  526. # - not CONNECT proxy must send absolute form URI
  527. # - most common is origin form URI
  528. if self.method == hdrs.METH_CONNECT:
  529. connect_host = self.url.raw_host
  530. assert connect_host is not None
  531. if helpers.is_ipv6_address(connect_host):
  532. connect_host = f"[{connect_host}]"
  533. path = f"{connect_host}:{self.url.port}"
  534. elif self.proxy and not self.is_ssl():
  535. path = str(self.url)
  536. else:
  537. path = self.url.raw_path
  538. if self.url.raw_query_string:
  539. path += "?" + self.url.raw_query_string
  540. protocol = conn.protocol
  541. assert protocol is not None
  542. writer = StreamWriter(
  543. protocol,
  544. self.loop,
  545. on_chunk_sent=functools.partial(
  546. self._on_chunk_request_sent, self.method, self.url
  547. ),
  548. )
  549. if self.compress:
  550. writer.enable_compression(self.compress)
  551. if self.chunked is not None:
  552. writer.enable_chunking()
  553. # set default content-type
  554. if (
  555. self.method in self.POST_METHODS
  556. and hdrs.CONTENT_TYPE not in self.skip_auto_headers
  557. and hdrs.CONTENT_TYPE not in self.headers
  558. ):
  559. self.headers[hdrs.CONTENT_TYPE] = "application/octet-stream"
  560. # set the connection header
  561. connection = self.headers.get(hdrs.CONNECTION)
  562. if not connection:
  563. if self.keep_alive():
  564. if self.version == HttpVersion10:
  565. connection = "keep-alive"
  566. else:
  567. if self.version == HttpVersion11:
  568. connection = "close"
  569. if connection is not None:
  570. self.headers[hdrs.CONNECTION] = connection
  571. # status + headers
  572. status_line = "{0} {1} HTTP/{2[0]}.{2[1]}".format(
  573. self.method, path, self.version
  574. )
  575. await writer.write_headers(status_line, self.headers)
  576. self._writer = self.loop.create_task(self.write_bytes(writer, conn))
  577. response_class = self.response_class
  578. assert response_class is not None
  579. self.response = response_class(
  580. self.method,
  581. self.original_url,
  582. writer=self._writer,
  583. continue100=self._continue,
  584. timer=self._timer,
  585. request_info=self.request_info,
  586. traces=self._traces,
  587. loop=self.loop,
  588. session=self._session,
  589. )
  590. return self.response
  591. async def close(self) -> None:
  592. if self._writer is not None:
  593. try:
  594. await self._writer
  595. finally:
  596. self._writer = None
  597. def terminate(self) -> None:
  598. if self._writer is not None:
  599. if not self.loop.is_closed():
  600. self._writer.cancel()
  601. self._writer = None
  602. async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
  603. for trace in self._traces:
  604. await trace.send_request_chunk_sent(method, url, chunk)
  605. class ClientResponse(HeadersMixin):
  606. # from the Status-Line of the response
  607. version = None # HTTP-Version
  608. status = None # type: int # Status-Code
  609. reason = None # Reason-Phrase
  610. content = None # type: StreamReader # Payload stream
  611. _headers = None # type: CIMultiDictProxy[str] # Response headers
  612. _raw_headers = None # type: RawHeaders # Response raw headers
  613. _connection = None # current connection
  614. _source_traceback = None
  615. # setted up by ClientRequest after ClientResponse object creation
  616. # post-init stage allows to not change ctor signature
  617. _closed = True # to allow __del__ for non-initialized properly response
  618. _released = False
  619. def __init__(
  620. self,
  621. method: str,
  622. url: URL,
  623. *,
  624. writer: "asyncio.Task[None]",
  625. continue100: Optional["asyncio.Future[bool]"],
  626. timer: BaseTimerContext,
  627. request_info: RequestInfo,
  628. traces: List["Trace"],
  629. loop: asyncio.AbstractEventLoop,
  630. session: "ClientSession",
  631. ) -> None:
  632. assert isinstance(url, URL)
  633. self.method = method
  634. self.cookies = SimpleCookie() # type: SimpleCookie[str]
  635. self._real_url = url
  636. self._url = url.with_fragment(None)
  637. self._body = None # type: Any
  638. self._writer = writer # type: Optional[asyncio.Task[None]]
  639. self._continue = continue100 # None by default
  640. self._closed = True
  641. self._history = () # type: Tuple[ClientResponse, ...]
  642. self._request_info = request_info
  643. self._timer = timer if timer is not None else TimerNoop()
  644. self._cache = {} # type: Dict[str, Any]
  645. self._traces = traces
  646. self._loop = loop
  647. # store a reference to session #1985
  648. self._session = session # type: Optional[ClientSession]
  649. if loop.get_debug():
  650. self._source_traceback = traceback.extract_stack(sys._getframe(1))
  651. @reify
  652. def url(self) -> URL:
  653. return self._url
  654. @reify
  655. def url_obj(self) -> URL:
  656. warnings.warn("Deprecated, use .url #1654", DeprecationWarning, stacklevel=2)
  657. return self._url
  658. @reify
  659. def real_url(self) -> URL:
  660. return self._real_url
  661. @reify
  662. def host(self) -> str:
  663. assert self._url.host is not None
  664. return self._url.host
  665. @reify
  666. def headers(self) -> "CIMultiDictProxy[str]":
  667. return self._headers
  668. @reify
  669. def raw_headers(self) -> RawHeaders:
  670. return self._raw_headers
  671. @reify
  672. def request_info(self) -> RequestInfo:
  673. return self._request_info
  674. @reify
  675. def content_disposition(self) -> Optional[ContentDisposition]:
  676. raw = self._headers.get(hdrs.CONTENT_DISPOSITION)
  677. if raw is None:
  678. return None
  679. disposition_type, params_dct = multipart.parse_content_disposition(raw)
  680. params = MappingProxyType(params_dct)
  681. filename = multipart.content_disposition_filename(params)
  682. return ContentDisposition(disposition_type, params, filename)
  683. def __del__(self, _warnings: Any = warnings) -> None:
  684. if self._closed:
  685. return
  686. if self._connection is not None:
  687. self._connection.release()
  688. self._cleanup_writer()
  689. if self._loop.get_debug():
  690. if PY_36:
  691. kwargs = {"source": self}
  692. else:
  693. kwargs = {}
  694. _warnings.warn(f"Unclosed response {self!r}", ResourceWarning, **kwargs)
  695. context = {"client_response": self, "message": "Unclosed response"}
  696. if self._source_traceback:
  697. context["source_traceback"] = self._source_traceback
  698. self._loop.call_exception_handler(context)
  699. def __repr__(self) -> str:
  700. out = io.StringIO()
  701. ascii_encodable_url = str(self.url)
  702. if self.reason:
  703. ascii_encodable_reason = self.reason.encode(
  704. "ascii", "backslashreplace"
  705. ).decode("ascii")
  706. else:
  707. ascii_encodable_reason = self.reason
  708. print(
  709. "<ClientResponse({}) [{} {}]>".format(
  710. ascii_encodable_url, self.status, ascii_encodable_reason
  711. ),
  712. file=out,
  713. )
  714. print(self.headers, file=out)
  715. return out.getvalue()
  716. @property
  717. def connection(self) -> Optional["Connection"]:
  718. return self._connection
  719. @reify
  720. def history(self) -> Tuple["ClientResponse", ...]:
  721. """A sequence of of responses, if redirects occurred."""
  722. return self._history
  723. @reify
  724. def links(self) -> "MultiDictProxy[MultiDictProxy[Union[str, URL]]]":
  725. links_str = ", ".join(self.headers.getall("link", []))
  726. if not links_str:
  727. return MultiDictProxy(MultiDict())
  728. links = MultiDict() # type: MultiDict[MultiDictProxy[Union[str, URL]]]
  729. for val in re.split(r",(?=\s*<)", links_str):
  730. match = re.match(r"\s*<(.*)>(.*)", val)
  731. if match is None: # pragma: no cover
  732. # the check exists to suppress mypy error
  733. continue
  734. url, params_str = match.groups()
  735. params = params_str.split(";")[1:]
  736. link = MultiDict() # type: MultiDict[Union[str, URL]]
  737. for param in params:
  738. match = re.match(r"^\s*(\S*)\s*=\s*(['\"]?)(.*?)(\2)\s*$", param, re.M)
  739. if match is None: # pragma: no cover
  740. # the check exists to suppress mypy error
  741. continue
  742. key, _, value, _ = match.groups()
  743. link.add(key, value)
  744. key = link.get("rel", url) # type: ignore
  745. link.add("url", self.url.join(URL(url)))
  746. links.add(key, MultiDictProxy(link))
  747. return MultiDictProxy(links)
  748. async def start(self, connection: "Connection") -> "ClientResponse":
  749. """Start response processing."""
  750. self._closed = False
  751. self._protocol = connection.protocol
  752. self._connection = connection
  753. with self._timer:
  754. while True:
  755. # read response
  756. try:
  757. message, payload = await self._protocol.read() # type: ignore
  758. except http.HttpProcessingError as exc:
  759. raise ClientResponseError(
  760. self.request_info,
  761. self.history,
  762. status=exc.code,
  763. message=exc.message,
  764. headers=exc.headers,
  765. ) from exc
  766. if message.code < 100 or message.code > 199 or message.code == 101:
  767. break
  768. if self._continue is not None:
  769. set_result(self._continue, True)
  770. self._continue = None
  771. # payload eof handler
  772. payload.on_eof(self._response_eof)
  773. # response status
  774. self.version = message.version
  775. self.status = message.code
  776. self.reason = message.reason
  777. # headers
  778. self._headers = message.headers # type is CIMultiDictProxy
  779. self._raw_headers = message.raw_headers # type is Tuple[bytes, bytes]
  780. # payload
  781. self.content = payload
  782. # cookies
  783. for hdr in self.headers.getall(hdrs.SET_COOKIE, ()):
  784. try:
  785. self.cookies.load(hdr)
  786. except CookieError as exc:
  787. client_logger.warning("Can not load response cookies: %s", exc)
  788. return self
  789. def _response_eof(self) -> None:
  790. if self._closed:
  791. return
  792. if self._connection is not None:
  793. # websocket, protocol could be None because
  794. # connection could be detached
  795. if (
  796. self._connection.protocol is not None
  797. and self._connection.protocol.upgraded
  798. ):
  799. return
  800. self._connection.release()
  801. self._connection = None
  802. self._closed = True
  803. self._cleanup_writer()
  804. @property
  805. def closed(self) -> bool:
  806. return self._closed
  807. def close(self) -> None:
  808. if not self._released:
  809. self._notify_content()
  810. if self._closed:
  811. return
  812. self._closed = True
  813. if self._loop is None or self._loop.is_closed():
  814. return
  815. if self._connection is not None:
  816. self._connection.close()
  817. self._connection = None
  818. self._cleanup_writer()
  819. def release(self) -> Any:
  820. if not self._released:
  821. self._notify_content()
  822. if self._closed:
  823. return noop()
  824. self._closed = True
  825. if self._connection is not None:
  826. self._connection.release()
  827. self._connection = None
  828. self._cleanup_writer()
  829. return noop()
  830. @property
  831. def ok(self) -> bool:
  832. """Returns ``True`` if ``status`` is less than ``400``, ``False`` if not.
  833. This is **not** a check for ``200 OK`` but a check that the response
  834. status is under 400.
  835. """
  836. try:
  837. self.raise_for_status()
  838. except ClientResponseError:
  839. return False
  840. return True
  841. def raise_for_status(self) -> None:
  842. if 400 <= self.status:
  843. # reason should always be not None for a started response
  844. assert self.reason is not None
  845. self.release()
  846. raise ClientResponseError(
  847. self.request_info,
  848. self.history,
  849. status=self.status,
  850. message=self.reason,
  851. headers=self.headers,
  852. )
  853. def _cleanup_writer(self) -> None:
  854. if self._writer is not None:
  855. self._writer.cancel()
  856. self._writer = None
  857. self._session = None
  858. def _notify_content(self) -> None:
  859. content = self.content
  860. if content and content.exception() is None:
  861. content.set_exception(ClientConnectionError("Connection closed"))
  862. self._released = True
  863. async def wait_for_close(self) -> None:
  864. if self._writer is not None:
  865. try:
  866. await self._writer
  867. finally:
  868. self._writer = None
  869. self.release()
  870. async def read(self) -> bytes:
  871. """Read response payload."""
  872. if self._body is None:
  873. try:
  874. self._body = await self.content.read()
  875. for trace in self._traces:
  876. await trace.send_response_chunk_received(
  877. self.method, self.url, self._body
  878. )
  879. except BaseException:
  880. self.close()
  881. raise
  882. elif self._released:
  883. raise ClientConnectionError("Connection closed")
  884. return self._body
  885. def get_encoding(self) -> str:
  886. ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
  887. mimetype = helpers.parse_mimetype(ctype)
  888. encoding = mimetype.parameters.get("charset")
  889. if encoding:
  890. try:
  891. codecs.lookup(encoding)
  892. except LookupError:
  893. encoding = None
  894. if not encoding:
  895. if mimetype.type == "application" and (
  896. mimetype.subtype == "json" or mimetype.subtype == "rdap"
  897. ):
  898. # RFC 7159 states that the default encoding is UTF-8.
  899. # RFC 7483 defines application/rdap+json
  900. encoding = "utf-8"
  901. elif self._body is None:
  902. raise RuntimeError(
  903. "Cannot guess the encoding of " "a not yet read body"
  904. )
  905. else:
  906. encoding = chardet.detect(self._body)["encoding"]
  907. if not encoding:
  908. encoding = "utf-8"
  909. return encoding
  910. async def text(self, encoding: Optional[str] = None, errors: str = "strict") -> str:
  911. """Read response payload and decode."""
  912. if self._body is None:
  913. await self.read()
  914. if encoding is None:
  915. encoding = self.get_encoding()
  916. return self._body.decode(encoding, errors=errors) # type: ignore
  917. async def json(
  918. self,
  919. *,
  920. encoding: Optional[str] = None,
  921. loads: JSONDecoder = DEFAULT_JSON_DECODER,
  922. content_type: Optional[str] = "application/json",
  923. ) -> Any:
  924. """Read and decodes JSON response."""
  925. if self._body is None:
  926. await self.read()
  927. if content_type:
  928. ctype = self.headers.get(hdrs.CONTENT_TYPE, "").lower()
  929. if not _is_expected_content_type(ctype, content_type):
  930. raise ContentTypeError(
  931. self.request_info,
  932. self.history,
  933. message=(
  934. "Attempt to decode JSON with " "unexpected mimetype: %s" % ctype
  935. ),
  936. headers=self.headers,
  937. )
  938. stripped = self._body.strip() # type: ignore
  939. if not stripped:
  940. return None
  941. if encoding is None:
  942. encoding = self.get_encoding()
  943. return loads(stripped.decode(encoding))
  944. async def __aenter__(self) -> "ClientResponse":
  945. return self
  946. async def __aexit__(
  947. self,
  948. exc_type: Optional[Type[BaseException]],
  949. exc_val: Optional[BaseException],
  950. exc_tb: Optional[TracebackType],
  951. ) -> None:
  952. # similar to _RequestContextManager, we do not need to check
  953. # for exceptions, response object can close connection
  954. # if state is broken
  955. self.release()