(self, options: FinalRequestOptions, *, retries_taken: int = 0)
| 429 | raise NotImplementedError() |
| 430 | |
| 431 | def _build_headers(self, options: FinalRequestOptions, *, retries_taken: int = 0) -> httpx.Headers: |
| 432 | custom_headers = options.headers or {} |
| 433 | headers_dict = _merge_mappings(self.default_headers, custom_headers) |
| 434 | self._validate_headers(headers_dict, custom_headers) |
| 435 | |
| 436 | # headers are case-insensitive while dictionaries are not. |
| 437 | headers = httpx.Headers(headers_dict) |
| 438 | |
| 439 | idempotency_header = self._idempotency_header |
| 440 | if idempotency_header and options.idempotency_key and idempotency_header not in headers: |
| 441 | headers[idempotency_header] = options.idempotency_key |
| 442 | |
| 443 | # Don't set these headers if they were already set or removed by the caller. We check |
| 444 | # `custom_headers`, which can contain `Omit()`, instead of `headers` to account for the removal case. |
| 445 | lower_custom_headers = [header.lower() for header in custom_headers] |
| 446 | if "x-stainless-retry-count" not in lower_custom_headers: |
| 447 | headers["x-stainless-retry-count"] = str(retries_taken) |
| 448 | if "x-stainless-read-timeout" not in lower_custom_headers: |
| 449 | timeout = self.timeout if isinstance(options.timeout, NotGiven) else options.timeout |
| 450 | if isinstance(timeout, Timeout): |
| 451 | timeout = timeout.read |
| 452 | if timeout is not None: |
| 453 | headers["x-stainless-read-timeout"] = str(timeout) |
| 454 | |
| 455 | return headers |
| 456 | |
| 457 | def _prepare_url(self, url: str) -> URL: |
| 458 | """ |
no test coverage detected