(self, request: Request)
| 48 | self._download_http = _http_handler.download_request |
| 49 | |
| 50 | async def download_request(self, request: Request) -> Response: |
| 51 | p = urlparse_cached(request) |
| 52 | scheme = "https" if request.meta.get("is_secure") else "http" |
| 53 | bucket = p.hostname |
| 54 | path = p.path + "?" + p.query if p.query else p.path |
| 55 | url = f"{scheme}://{bucket}.s3.amazonaws.com{path}" |
| 56 | if self.anon: |
| 57 | request = request.replace(url=url) |
| 58 | else: |
| 59 | import botocore.awsrequest # noqa: PLC0415 |
| 60 | |
| 61 | awsrequest = botocore.awsrequest.AWSRequest( |
| 62 | method=request.method, |
| 63 | url=f"{scheme}://s3.amazonaws.com/{bucket}{path}", |
| 64 | headers=cast("Mapping[str, Any]", request.headers.to_unicode_dict()), |
| 65 | data=request.body, |
| 66 | ) |
| 67 | assert self._signer |
| 68 | self._signer.add_auth(awsrequest) |
| 69 | request = request.replace(url=url, headers=awsrequest.headers.items()) |
| 70 | return await self._download_http(request) |
nothing calls this directly
no test coverage detected