from __future__ import annotations

import sys

import json

import time

import uuid

import email

import asyncio

import inspect

import logging

import secrets

import weakref

import platform

import warnings

import threading

import email.utils

from types import TracebackType

from random import random

from typing import (

TYPE_CHECKING,

Any,

Dict,

Type,

Union,

Generic,

Mapping,

TypeVar,

Callable,

Iterable,

Iterator,

Optional,

Generator,

AsyncIterator,

cast,

overload,

)

from typing_extensions import Literal, override, get_origin

import anyio

import httpx

import distro

import pydantic

from httpx import URL

from pydantic import PrivateAttr

from . import _exceptions

from ._qs import Querystring

from ._files import to_httpx_files, async_to_httpx_files

from ._types import (

Body,

Omit,

Query,

Headers,

Timeout,

NotGiven,

ResponseT,

AnyMapping,

PostParser,

BinaryTypes,

RequestFiles,

HttpxSendArgs,

RequestOptions,

AsyncBinaryTypes,

HttpxRequestFiles,

ModelBuilderProtocol,

not_given,

)

from ._utils import is_dict, is_list, asyncify, is_given, lru_cache, is_mapping

from ._compat import PYDANTIC_V1, model_copy, model_dump

from ._models import GenericModel, FinalRequestOptions, validate_type, construct_type

from ._response import (

APIResponse,

BaseAPIResponse,

AsyncAPIResponse,

extract_response_type,

)

from ._constants import (

DEFAULT_TIMEOUT,

MAX_RETRY_DELAY,

DEFAULT_MAX_RETRIES,

INITIAL_RETRY_DELAY,

RAW_RESPONSE_HEADER,

DEFAULT_API_POOL_SHARDS,

OVERRIDE_CAST_TO_HEADER,

DEFAULT_CONNECTION_LIMITS,

DEFAULT_TRANSFER_POOL_SHARDS,

DEFAULT_BACKGROUND_POOL_SHARDS,

)

from ._streaming import Stream, SSEDecoder, AsyncStream, SSEBytesDecoder

from ._exceptions import (

APIStatusError,

APITimeoutError,

APIConnectionError,

APIResponseValidationError,

)

from ._utils._json import openapi_dumps

log: logging.Logger = logging.getLogger(__name__)

# Shared HTTP transport state. We share transports (connection pools) rather

# than full httpx clients so each SDK instance keeps its own cookie jar and

# mutable client state. Refcounted wrappers close the real transport only

# when the last user releases it.

# The async transport is keyed by event loop because connections bind to the

# loop that created them and cannot be reused across asyncio.run() calls.

#

# httpx/httpcore use a single H2 connection per transport, so each workload

# category uses a sharded Registry of SharedTransport instances. Per-client

# round-robin spreads requests across shards.

_pool_lock = threading.Lock()

class _SharedTransport(httpx.BaseTransport):

"""Refcounted wrapper: delegates to a real transport, closes it when refcount hits 0."""

def __init__(self, transport: httpx.BaseTransport) -> None:

self._transport = transport

self._refcount = 1

self._lock = threading.Lock()

@property

def refcount(self) -> int:

return self._refcount

def acquire(self) -> bool:

with self._lock:

if self._refcount <= 0:

return False

self._refcount += 1

return True

@override

def handle_request(self, request: httpx.Request) -> httpx.Response:

return self._transport.handle_request(request)

@override

def close(self) -> None:

should_close = False

with self._lock:

self._refcount -= 1

if self._refcount <= 0:

should_close = True

if should_close:

self._transport.close()

class Registry:

"""Process-global map of shard index → shared H2 transport."""

def __init__(self) -> None:

self._shards: dict[int, _SharedTransport] = {}

def acquire(self, shard: int) -> _SharedTransport:

with _pool_lock:

existing = self._shards.get(shard)

if existing is not None and existing.acquire():

return existing

transport = _SharedTransport(

httpx.HTTPTransport(limits=DEFAULT_CONNECTION_LIMITS, http2=True),

)

self._shards[shard] = transport

return transport

def take_all(self) -> list[_SharedTransport]:

"""Test-only: remove and return all transports for fixture cleanup."""

with _pool_lock:

values = list(self._shards.values())

self._shards.clear()

return values

def shard_ids(self) -> set[int]:

"""Test-only: which shard indexes currently hold a shared transport."""

with _pool_lock:

return set(self._shards)

class _SharedAsyncTransport(httpx.AsyncBaseTransport):

"""Async refcounted wrapper: delegates to a real async transport."""

def __init__(self, transport: httpx.AsyncBaseTransport) -> None:

self._transport = transport

self._refcount = 1

self._lock = threading.Lock()

@property

def refcount(self) -> int:

return self._refcount

def acquire(self) -> bool:

with self._lock:

if self._refcount <= 0:

return False

self._refcount += 1

return True

@override

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:

return await self._transport.handle_async_request(request)

@override

async def aclose(self) -> None:

should_close = False

with self._lock:

self._refcount -= 1

if self._refcount <= 0:

should_close = True

if should_close:

await self._transport.aclose()

class Registry:

"""Per-event-loop map of shard index → shared async H2 transport."""

def __init__(self) -> None:

self._by_loop: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[int, _SharedAsyncTransport]] = (

weakref.WeakKeyDictionary()

)

def acquire(self, loop: asyncio.AbstractEventLoop, shard: int) -> _SharedAsyncTransport:

with _pool_lock:

bucket = self._by_loop.get(loop)

if bucket is None:

bucket = {}

self._by_loop[loop] = bucket

existing = bucket.get(shard)

if existing is not None and existing.acquire():

return existing

transport = _SharedAsyncTransport(

httpx.AsyncHTTPTransport(limits=DEFAULT_CONNECTION_LIMITS, http2=True),

)

bucket[shard] = transport

return transport

def clear(self) -> None:

"""Test-only: drop all per-loop shard maps (does not close transports)."""

with _pool_lock:

self._by_loop.clear()

# Process-global sharded registries (one ≈ H2 connection per shard index).

_shared_sync_api_transports = _SharedTransport.Registry()

_shared_sync_background_transports = _SharedTransport.Registry()

_shared_sync_transfer_transports = _SharedTransport.Registry()

_shared_async_api_transports = _SharedAsyncTransport.Registry()

_shared_async_background_transports = _SharedAsyncTransport.Registry()

_shared_async_transfer_transports = _SharedAsyncTransport.Registry()

_BACKGROUND_PATH_SUFFIXES = ("/wait_for_status",)

_TRANSFER_PATH_SUFFIXES = ("/upload_file", "/download_file")

def _is_background_path(path: str) -> bool:

return path.endswith(_BACKGROUND_PATH_SUFFIXES)

def _is_transfer_path(path: str) -> bool:

return path.endswith(_TRANSFER_PATH_SUFFIXES)

class _SyncClientPool:

"""Per-SDK-client round-robin over sharded httpx.Client wrappers."""

def __init__(

self,

*,

shards: int,

shared: bool,

registry: _SharedTransport.Registry | None,

make_client: Callable[[httpx.BaseTransport | None], httpx.Client],

) -> None:

if shards < 1:

raise ValueError("shards must be >= 1")

self.shards = shards

self._shared = shared

self._registry = registry

self._make_client = make_client

self._clients: dict[int, httpx.Client] = {}

self._lock = threading.Lock()

self._next = secrets.randbelow(shards)

def ensure(self, shard: int) -> httpx.Client:

existing = self._clients.get(shard)

if existing is not None:

return existing

with self._lock:

existing = self._clients.get(shard)

if existing is not None:

return existing

transport: httpx.BaseTransport | None = None

if self._shared and self._registry is not None:

transport = self._registry.acquire(shard)

client = self._make_client(transport)

self._clients[shard] = client

return client

def next_client(self) -> httpx.Client:

with self._lock:

shard = self._next % self.shards

self._next += 1

return self.ensure(shard)

def close(self) -> None:

for client in self._clients.values():

client.close()

self._clients.clear()

class _AsyncClientPool:

"""Per-SDK-client round-robin over sharded httpx.AsyncClient wrappers."""

def __init__(

self,

*,

shards: int,

shared: bool,

registry: _SharedAsyncTransport.Registry | None,

make_client: Callable[[httpx.AsyncBaseTransport | None], httpx.AsyncClient],

) -> None:

if shards < 1:

raise ValueError("shards must be >= 1")

self.shards = shards

self._shared = shared

self._registry = registry

self._make_client = make_client

self._clients: dict[int, httpx.AsyncClient] = {}

self._next = secrets.randbelow(shards)

def ensure(self, shard: int) -> httpx.AsyncClient:

existing = self._clients.get(shard)

if existing is not None:

return existing

transport: httpx.AsyncBaseTransport | None = None

if self._shared and self._registry is not None:

try:

loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()

except RuntimeError:

loop = None

if loop is not None:

transport = self._registry.acquire(loop, shard)

client = self._make_client(transport)

self._clients[shard] = client

return client

def next_client(self) -> httpx.AsyncClient:

# Single-threaded event loop: counter bump needs no lock when there is no await.

shard = self._next % self.shards

self._next += 1

return self.ensure(shard)

async def aclose(self) -> None:

for client in self._clients.values():

await client.aclose()

self._clients.clear()

# TODO: make base page type vars covariant

SyncPageT = TypeVar("SyncPageT", bound="BaseSyncPage[Any]")

AsyncPageT = TypeVar("AsyncPageT", bound="BaseAsyncPage[Any]")

_T = TypeVar("_T")

_T_co = TypeVar("_T_co", covariant=True)

_StreamT = TypeVar("_StreamT", bound=Stream[Any])

_AsyncStreamT = TypeVar("_AsyncStreamT", bound=AsyncStream[Any])

if TYPE_CHECKING:

from httpx._config import (

DEFAULT_TIMEOUT_CONFIG, # pyright: ignore[reportPrivateImportUsage]

)

HTTPX_DEFAULT_TIMEOUT = DEFAULT_TIMEOUT_CONFIG

else:

try:

from httpx._config import DEFAULT_TIMEOUT_CONFIG as HTTPX_DEFAULT_TIMEOUT

except ImportError:

# taken from https://github.com/encode/httpx/blob/3ba5fe0d7ac70222590e759c31442b1cab263791/httpx/_config.py#L366

HTTPX_DEFAULT_TIMEOUT = Timeout(5.0)

class PageInfo:

"""Stores the necessary information to build the request to retrieve the next page.

Either `url` or `params` must be set.

"""

url: URL | NotGiven

params: Query | NotGiven

json: Body | NotGiven

@overload

def __init__(

self,

*,

url: URL,

) -> None: ...

@overload

def __init__(

self,

*,

params: Query,

) -> None: ...

@overload

def __init__(

self,

*,

json: Body,

) -> None: ...

def __init__(

self,

*,

url: URL | NotGiven = not_given,

json: Body | NotGiven = not_given,

params: Query | NotGiven = not_given,

) -> None:

self.url = url

self.json = json

self.params = params

@override

def __repr__(self) -> str:

if self.url:

return f"{self.__class__.__name__}(url={self.url})"

if self.json:

return f"{self.__class__.__name__}(json={self.json})"

return f"{self.__class__.__name__}(params={self.params})"

class BasePage(GenericModel, Generic[_T]):

"""

Defines the core interface for pagination.

Type Args:

ModelT: The pydantic model that represents an item in the response.

Methods:

has_next_page(): Check if there is another page available

next_page_info(): Get the necessary information to make a request for the next page

"""

_options: FinalRequestOptions = PrivateAttr()

_model: Type[_T] = PrivateAttr()

def has_next_page(self) -> bool:

items = self._get_page_items()

if not items:

return False

return self.next_page_info() is not None

def next_page_info(self) -> Optional[PageInfo]: ...

def _get_page_items(self) -> Iterable[_T]: # type: ignore[empty-body]

...

def _params_from_url(self, url: URL) -> httpx.QueryParams:

# TODO: do we have to preprocess params here?

return httpx.QueryParams(cast(Any, self._options.params)).merge(url.params)

def _info_to_options(self, info: PageInfo) -> FinalRequestOptions:

options = model_copy(self._options)

options._strip_raw_response_header()

if not isinstance(info.params, NotGiven):

options.params = {**options.params, **info.params}

return options

if not isinstance(info.url, NotGiven):

params = self._params_from_url(info.url)

url = info.url.copy_with(params=params)

options.params = dict(url.params)

options.url = str(url)

return options

if not isinstance(info.json, NotGiven):

if not is_mapping(info.json):

raise TypeError("Pagination is only supported with mappings")

if not options.json_data:

options.json_data = {**info.json}

else:

if not is_mapping(options.json_data):

raise TypeError("Pagination is only supported with mappings")

options.json_data = {**options.json_data, **info.json}

return options

raise ValueError("Unexpected PageInfo state")

class BaseSyncPage(BasePage[_T], Generic[_T]):

_client: SyncAPIClient = pydantic.PrivateAttr()

def _set_private_attributes(

self,

client: SyncAPIClient,

model: Type[_T],

options: FinalRequestOptions,

) -> None:

if (not PYDANTIC_V1) and getattr(self, "__pydantic_private__", None) is None:

self.__pydantic_private__ = {}

self._model = model

self._client = client

self._options = options

# Pydantic uses a custom `__iter__` method to support casting BaseModels

# to dictionaries. e.g. dict(model).

# As we want to support `for item in page`, this is inherently incompatible

# with the default pydantic behaviour. It is not possible to support both

# use cases at once. Fortunately, this is not a big deal as all other pydantic

# methods should continue to work as expected as there is an alternative method

# to cast a model to a dictionary, model.dict(), which is used internally

# by pydantic.

def __iter__(self) -> Iterator[_T]: # type: ignore

for page in self.iter_pages():

for item in page._get_page_items():

yield item

def iter_pages(self: SyncPageT) -> Iterator[SyncPageT]:

page = self

while True:

yield page

if page.has_next_page():

page = page.get_next_page()

else:

return

def get_next_page(self: SyncPageT) -> SyncPageT:

info = self.next_page_info()

if not info:

raise RuntimeError(

"No next page expected; please check `.has_next_page()` before calling `.get_next_page()`."

)

options = self._info_to_options(info)

return self._client._request_api_list(self._model, page=self.__class__, options=options)

class AsyncPaginator(Generic[_T, AsyncPageT]):

def __init__(

self,

client: AsyncAPIClient,

options: FinalRequestOptions,

page_cls: Type[AsyncPageT],

model: Type[_T],

) -> None:

self._model = model

self._client = client

self._options = options

self._page_cls = page_cls

def __await__(self) -> Generator[Any, None, AsyncPageT]:

return self._get_page().__await__()

async def _get_page(self) -> AsyncPageT:

def _parser(resp: AsyncPageT) -> AsyncPageT:

resp._set_private_attributes(

model=self._model,

options=self._options,

client=self._client,

)

return resp

self._options.post_parser = _parser

return await self._client.request(self._page_cls, self._options)

async def __aiter__(self) -> AsyncIterator[_T]:

# https://github.com/microsoft/pyright/issues/3464

page = cast(

AsyncPageT,

await self, # type: ignore

)

async for item in page:

yield item

class BaseAsyncPage(BasePage[_T], Generic[_T]):

_client: AsyncAPIClient = pydantic.PrivateAttr()

def _set_private_attributes(

self,

model: Type[_T],

client: AsyncAPIClient,

options: FinalRequestOptions,

) -> None:

if (not PYDANTIC_V1) and getattr(self, "__pydantic_private__", None) is None:

self.__pydantic_private__ = {}

self._model = model

self._client = client

self._options = options

async def __aiter__(self) -> AsyncIterator[_T]:

async for page in self.iter_pages():

for item in page._get_page_items():

yield item

async def iter_pages(self: AsyncPageT) -> AsyncIterator[AsyncPageT]:

page = self

while True:

yield page

if page.has_next_page():

page = await page.get_next_page()

else:

return

async def get_next_page(self: AsyncPageT) -> AsyncPageT:

info = self.next_page_info()

if not info:

raise RuntimeError(

"No next page expected; please check `.has_next_page()` before calling `.get_next_page()`."

)

options = self._info_to_options(info)

return await self._client._request_api_list(self._model, page=self.__class__, options=options)

_HttpxClientT = TypeVar("_HttpxClientT", bound=Union[httpx.Client, httpx.AsyncClient])

_DefaultStreamT = TypeVar("_DefaultStreamT", bound=Union[Stream[Any], AsyncStream[Any]])

class BaseClient(Generic[_HttpxClientT, _DefaultStreamT]):

_client: _HttpxClientT

_version: str

_base_url: URL

max_retries: int

timeout: Union[float, Timeout, None]

_strict_response_validation: bool

_idempotency_header: str | None

_default_stream_cls: type[_DefaultStreamT] | None = None

def __init__(

self,

*,

version: str,

base_url: str | URL,

_strict_response_validation: bool,

max_retries: int = DEFAULT_MAX_RETRIES,

timeout: float | Timeout | None = DEFAULT_TIMEOUT,

custom_headers: Mapping[str, str] | None = None,

custom_query: Mapping[str, object] | None = None,

) -> None:

self._version = version

self._base_url = self._enforce_trailing_slash(URL(base_url))

self.max_retries = max_retries

self.timeout = timeout

self._custom_headers = custom_headers or {}

self._custom_query = custom_query or {}

self._strict_response_validation = _strict_response_validation

self._idempotency_header = None

self._platform: Platform | None = None

if max_retries is None: # pyright: ignore[reportUnnecessaryComparison]

raise TypeError(

"max_retries cannot be None. If you want to disable retries, pass `0`; if you want unlimited retries, pass `math.inf` or a very high number; if you want the default behavior, pass `runloop_api_client.DEFAULT_MAX_RETRIES`"

)

def _enforce_trailing_slash(self, url: URL) -> URL:

if url.raw_path.endswith(b"/"):

return url

return url.copy_with(raw_path=url.raw_path + b"/")

def _make_status_error_from_response(

self,

response: httpx.Response,

) -> APIStatusError:

if response.is_closed and not response.is_stream_consumed:

# We can't read the response body as it has been closed

# before it was read. This can happen if an event hook

# raises a status error.

body = None

err_msg = f"Error code: {response.status_code}"

else:

err_text = response.text.strip()

body = err_text

try:

body = json.loads(err_text)

err_msg = f"Error code: {response.status_code} - {body}"

except Exception:

err_msg = err_text or f"Error code: {response.status_code}"

return self._make_status_error(err_msg, body=body, response=response)

def _make_status_error(

self,

err_msg: str,

*,

body: object,

response: httpx.Response,

) -> _exceptions.APIStatusError:

raise NotImplementedError()

def _build_headers(self, options: FinalRequestOptions, *, retries_taken: int = 0) -> httpx.Headers:

custom_headers = options.headers or {}

headers_dict = _merge_mappings(self.default_headers, custom_headers)

self._validate_headers(headers_dict, custom_headers)

# headers are case-insensitive while dictionaries are not.

headers = httpx.Headers(headers_dict)

idempotency_header = self._idempotency_header

if idempotency_header and options.idempotency_key and idempotency_header not in headers:

headers[idempotency_header] = options.idempotency_key

# Don't set these headers if they were already set or removed by the caller. We check

# `custom_headers`, which can contain `Omit()`, instead of `headers` to account for the removal case.

lower_custom_headers = [header.lower() for header in custom_headers]

if "x-stainless-retry-count" not in lower_custom_headers:

headers["x-stainless-retry-count"] = str(retries_taken)

if "x-stainless-read-timeout" not in lower_custom_headers:

timeout = self.timeout if isinstance(options.timeout, NotGiven) else options.timeout

if isinstance(timeout, Timeout):

timeout = timeout.read

if timeout is not None:

headers["x-stainless-read-timeout"] = str(timeout)

return headers

def _prepare_url(self, url: str) -> URL:

"""

Merge a URL argument together with any 'base_url' on the client,

to create the URL used for the outgoing request.

"""

# Copied from httpx's `_merge_url` method.

merge_url = URL(url)

if merge_url.is_relative_url:

merge_raw_path = self.base_url.raw_path + merge_url.raw_path.lstrip(b"/")

return self.base_url.copy_with(raw_path=merge_raw_path)

return merge_url

def _make_sse_decoder(self) -> SSEDecoder | SSEBytesDecoder:

return SSEDecoder()

def _build_request(

self,

options: FinalRequestOptions,

*,

retries_taken: int = 0,

) -> httpx.Request:

if log.isEnabledFor(logging.DEBUG):

log.debug(

"Request options: %s",

model_dump(

options,

exclude_unset=True,

# Pydantic v1 can't dump every type we support in content, so we exclude it for now.

exclude={

"content",

}

if PYDANTIC_V1

else {},

),

)

kwargs: dict[str, Any] = {}

json_data = options.json_data

if options.extra_json is not None:

if json_data is None:

json_data = cast(Body, options.extra_json)

elif is_mapping(json_data):

json_data = _merge_mappings(json_data, options.extra_json)

else:

raise RuntimeError(f"Unexpected JSON data type, {type(json_data)}, cannot merge with `extra_body`")

headers = self._build_headers(options, retries_taken=retries_taken)

params = _merge_mappings(self.default_query, options.params)

content_type = headers.get("Content-Type")

files = options.files

# If the given Content-Type header is multipart/form-data then it

# has to be removed so that httpx can generate the header with

# additional information for us as it has to be in this form

# for the server to be able to correctly parse the request:

# multipart/form-data; boundary=---abc--

if content_type is not None and content_type.startswith("multipart/form-data"):

if "boundary" not in content_type:

# only remove the header if the boundary hasn't been explicitly set

# as the caller doesn't want httpx to come up with their own boundary

headers.pop("Content-Type")

# As we are now sending multipart/form-data instead of application/json

# we need to tell httpx to use it, https://www.python-httpx.org/advanced/clients/#multipart-file-encoding

if json_data:

if not is_dict(json_data):

raise TypeError(

f"Expected query input to be a dictionary for multipart requests but got {type(json_data)} instead."

)

kwargs["data"] = self._serialize_multipartform(json_data)

# httpx determines whether or not to send a "multipart/form-data"

# request based on the truthiness of the "files" argument.

# This gets around that issue by generating a dict value that

# evaluates to true.

#

# https://github.com/encode/httpx/discussions/2399#discussioncomment-3814186

if not files:

files = cast(HttpxRequestFiles, ForceMultipartDict())

prepared_url = self._prepare_url(options.url)

# preserve hard-coded query params from the url

if params and prepared_url.query:

params = {**dict(prepared_url.params.items()), **params}

prepared_url = prepared_url.copy_with(raw_path=prepared_url.raw_path.split(b"?", 1)[0])

if "_" in prepared_url.host:

# work around https://github.com/encode/httpx/discussions/2880

kwargs["extensions"] = {"sni_hostname": prepared_url.host.replace("_", "-")}

is_body_allowed = options.method.lower() != "get"

if is_body_allowed:

if options.content is not None and json_data is not None:

raise TypeError("Passing both `content` and `json_data` is not supported")

if options.content is not None and files is not None:

raise TypeError("Passing both `content` and `files` is not supported")

if options.content is not None:

kwargs["content"] = options.content

elif isinstance(json_data, bytes):

kwargs["content"] = json_data

elif not files:

# Don't set content when JSON is sent as multipart/form-data,

# since httpx's content param overrides other body arguments

kwargs["content"] = openapi_dumps(json_data) if is_given(json_data) and json_data is not None else None

kwargs["files"] = files

else:

headers.pop("Content-Type", None)

kwargs.pop("data", None)

# TODO: report this error to httpx

return self._client.build_request( # pyright: ignore[reportUnknownMemberType]

headers=headers,

timeout=self.timeout if isinstance(options.timeout, NotGiven) else options.timeout,

method=options.method,

url=prepared_url,

# the `Query` type that we use is incompatible with qs'

# `Params` type as it needs to be typed as `Mapping[str, object]`

# so that passing a `TypedDict` doesn't cause an error.

# https://github.com/microsoft/pyright/issues/3526#event-6715453066

params=self.qs.stringify(cast(Mapping[str, Any], params)) if params else None,

**kwargs,

)

def _serialize_multipartform(self, data: Mapping[object, object]) -> dict[str, object]:

items = self.qs.stringify_items(

# TODO: type ignore is required as stringify_items is well typed but we can't be

# well typed without heavy validation.

data, # type: ignore

array_format="brackets",

)

serialized: dict[str, object] = {}

for key, value in items:

existing = serialized.get(key)

if not existing:

serialized[key] = value

continue

# If a value has already been set for this key then that

# means we're sending data like `array[]=[1, 2, 3]` and we

# need to tell httpx that we want to send multiple values with

# the same key which is done by using a list or a tuple.

#

# Note: 2d arrays should never result in the same key at both

# levels so it's safe to assume that if the value is a list,

# it was because we changed it to be a list.

if is_list(existing):

existing.append(value)

else:

serialized[key] = [existing, value]

return serialized

def _maybe_override_cast_to(self, cast_to: type[ResponseT], options: FinalRequestOptions) -> type[ResponseT]:

if not is_given(options.headers):

return cast_to

# make a copy of the headers so we don't mutate user-input

headers = dict(options.headers)

# we internally support defining a temporary header to override the

# default `cast_to` type for use with `.with_raw_response` and `.with_streaming_response`

# see _response.py for implementation details

override_cast_to = headers.pop(OVERRIDE_CAST_TO_HEADER, not_given)

if is_given(override_cast_to):

options.headers = headers

return cast(Type[ResponseT], override_cast_to)

return cast_to

def _should_stream_response_body(self, request: httpx.Request) -> bool:

return request.headers.get(RAW_RESPONSE_HEADER) == "stream" # type: ignore[no-any-return]

def _process_response_data(

self,

*,

data: object,

cast_to: type[ResponseT],

response: httpx.Response,

) -> ResponseT:

if data is None:

return cast(ResponseT, None)

if cast_to is object:

return cast(ResponseT, data)

try:

if inspect.isclass(cast_to) and issubclass(cast_to, ModelBuilderProtocol):

return cast(ResponseT, cast_to.build(response=response, data=data))

if self._strict_response_validation:

return cast(ResponseT, validate_type(type_=cast_to, value=data))

return cast(ResponseT, construct_type(type_=cast_to, value=data))

except pydantic.ValidationError as err:

raise APIResponseValidationError(response=response, body=data) from err

@property

def qs(self) -> Querystring:

return Querystring()

@property

def custom_auth(self) -> httpx.Auth | None:

return None

@property

def auth_headers(self) -> dict[str, str]:

return {}

@property

def default_headers(self) -> dict[str, str | Omit]:

return {

"Accept": "application/json",

"Content-Type": "application/json",

"User-Agent": self.user_agent,

**self.platform_headers(),

**self.auth_headers,

**self._custom_headers,

}

@property

def default_query(self) -> dict[str, object]:

return {

**self._custom_query,

}

def _validate_headers(

self,

headers: Headers, # noqa: ARG002

custom_headers: Headers, # noqa: ARG002

) -> None:

"""Validate the given default headers and custom headers.

Does nothing by default.

"""

return

@property

def user_agent(self) -> str:

return f"{self.__class__.__name__}/Python {self._version}"

@property

def base_url(self) -> URL:

return self._base_url

@base_url.setter

def base_url(self, url: URL | str) -> None:

self._base_url = self._enforce_trailing_slash(url if isinstance(url, URL) else URL(url))

def platform_headers(self) -> Dict[str, str]:

# the actual implementation is in a separate `lru_cache` decorated

# function because adding `lru_cache` to methods will leak memory

# https://github.com/python/cpython/issues/88476

return platform_headers(self._version, platform=self._platform)

def _parse_retry_after_header(self, response_headers: Optional[httpx.Headers] = None) -> float | None:

"""Returns a float of the number of seconds (not milliseconds) to wait after retrying, or None if unspecified.

About the Retry-After header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After

See also https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After#syntax

"""

if response_headers is None:

return None

# First, try the non-standard `retry-after-ms` header for milliseconds,

# which is more precise than integer-seconds `retry-after`

try:

retry_ms_header = response_headers.get("retry-after-ms", None)

return float(retry_ms_header) / 1000

except (TypeError, ValueError):

pass

# Next, try parsing `retry-after` header as seconds (allowing nonstandard floats).

retry_header = response_headers.get("retry-after")

try:

# note: the spec indicates that this should only ever be an integer

# but if someone sends a float there's no reason for us to not respect it

return float(retry_header)