(
self,
first_line: Union[httputil.ResponseStartLine, httputil.RequestStartLine],
headers: httputil.HTTPHeaders,
)
| 586 | self._handle_exception(*sys.exc_info()) |
| 587 | |
| 588 | async def headers_received( |
| 589 | self, |
| 590 | first_line: Union[httputil.ResponseStartLine, httputil.RequestStartLine], |
| 591 | headers: httputil.HTTPHeaders, |
| 592 | ) -> None: |
| 593 | assert isinstance(first_line, httputil.ResponseStartLine) |
| 594 | if self.request.expect_100_continue and first_line.code == 100: |
| 595 | await self._write_body(False) |
| 596 | return |
| 597 | self.code = first_line.code |
| 598 | self.reason = first_line.reason |
| 599 | self.headers = headers |
| 600 | |
| 601 | if self._should_follow_redirect(): |
| 602 | return |
| 603 | |
| 604 | if self.request.header_callback is not None: |
| 605 | # Reassemble the start line. |
| 606 | self.request.header_callback("%s %s %s\r\n" % first_line) |
| 607 | for k, v in self.headers.get_all(): |
| 608 | self.request.header_callback(f"{k}: {v}\r\n") |
| 609 | self.request.header_callback("\r\n") |
| 610 | |
| 611 | def _should_follow_redirect(self) -> bool: |
| 612 | if self.request.follow_redirects: |
nothing calls this directly
no test coverage detected