(self, request: Request, send: ASGISendCallable)
| 104 | ) |
| 105 | |
| 106 | async def handle_request(self, request: Request, send: ASGISendCallable) -> None: |
| 107 | try: |
| 108 | response = await self.app.handle_request(request) |
| 109 | except Exception as error: |
| 110 | response = await _handle_exception(self.app, error) |
| 111 | |
| 112 | if isinstance(response, Response) and response.timeout != Ellipsis: |
| 113 | timeout = cast(Optional[float], response.timeout) |
| 114 | else: |
| 115 | timeout = self.app.config["RESPONSE_TIMEOUT"] |
| 116 | try: |
| 117 | await asyncio.wait_for(self._send_response(send, response), timeout=timeout) |
| 118 | except asyncio.TimeoutError: |
| 119 | pass |
| 120 | |
| 121 | async def _send_response( |
| 122 | self, send: ASGISendCallable, response: ResponseTypes |
no test coverage detected