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