(self, options: FinalRequestOptions, *, retries_taken: int = 0)
| 454 | return None |
| 455 | |
| 456 | def _build_headers(self, options: FinalRequestOptions, *, retries_taken: int = 0) -> httpx.Headers: |
| 457 | custom_headers = options.headers or {} |
| 458 | headers_dict = _merge_mappings({**self._auth_headers(options.security), **self.default_headers}, custom_headers) |
| 459 | self._validate_headers(headers_dict, custom_headers) |
| 460 | |
| 461 | # headers are case-insensitive while dictionaries are not. |
| 462 | headers = httpx.Headers(headers_dict) |
| 463 | |
| 464 | idempotency_header = self._idempotency_header |
| 465 | if idempotency_header and options.idempotency_key and idempotency_header not in headers: |
| 466 | headers[idempotency_header] = options.idempotency_key |
| 467 | |
| 468 | # Don't set these headers if they were already set or removed by the caller. We check |
| 469 | # `custom_headers`, which can contain `Omit()`, instead of `headers` to account for the removal case. |
| 470 | lower_custom_headers = [header.lower() for header in custom_headers] |
| 471 | if "x-stainless-retry-count" not in lower_custom_headers: |
| 472 | headers["x-stainless-retry-count"] = str(retries_taken) |
| 473 | if "x-stainless-read-timeout" not in lower_custom_headers: |
| 474 | timeout = self.timeout if isinstance(options.timeout, NotGiven) else options.timeout |
| 475 | if isinstance(timeout, Timeout): |
| 476 | timeout = timeout.read |
| 477 | if timeout is not None: |
| 478 | headers["x-stainless-read-timeout"] = str(timeout) |
| 479 | |
| 480 | return headers |
| 481 | |
| 482 | def _prepare_url(self, url: str) -> URL: |
| 483 | """ |
no test coverage detected