(self, request: httpx.Request, *, auth: BedrockAwsAuth, body: bytes)
| 153 | return _body_for_signing(request) |
| 154 | |
| 155 | def _sign(self, request: httpx.Request, *, auth: BedrockAwsAuth, body: bytes) -> None: |
| 156 | for header in _AWS_SIGNING_HEADERS: |
| 157 | request.headers.pop(header, None) |
| 158 | |
| 159 | signed_headers = auth.sign( |
| 160 | method=request.method, |
| 161 | url=str(request.url), |
| 162 | headers=dict(request.headers), |
| 163 | body=body, |
| 164 | ) |
| 165 | request.headers.clear() |
| 166 | request.headers.update(signed_headers) |
| 167 | |
| 168 | def prepare_request(self, request: httpx.Request) -> None: |
| 169 | body = self._validate_request(request) |
no test coverage detected