GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

openai/openai-python

Public

mirrored from https://github.com/openai/openai-pythonAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
646fff04dbe9458f1c46753031546ccf6415eeef

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

openai-python/src/openai

src/openai/_base_client.py

2049lines · modecode

RawDownload
Latest commit unavailable.
1from __future__ import annotations
2
3import json
4import time
5import uuid
6import email
7import asyncio
8import inspect
9import logging
10import platform
11import warnings
12import email.utils
13from types import TracebackType
14from random import random
15from typing import (
16 TYPE_CHECKING,
17 Any,
18 Dict,
19 Type,
20 Union,
21 Generic,
22 Mapping,
23 TypeVar,
24 Iterable,
25 Iterator,
26 Optional,
27 Generator,
28 AsyncIterator,
29 cast,
30 overload,
31)
32from typing_extensions import Literal, override, get_origin
33
34import anyio
35import httpx
36import distro
37import pydantic
38from httpx import URL, Limits
39from pydantic import PrivateAttr
40
41from . import _exceptions
42from ._qs import Querystring
43from ._files import to_httpx_files, async_to_httpx_files
44from ._types import (
45 NOT_GIVEN,
46 Body,
47 Omit,
48 Query,
49 Headers,
50 Timeout,
51 NotGiven,
52 ResponseT,
53 Transport,
54 AnyMapping,
55 PostParser,
56 ProxiesTypes,
57 RequestFiles,
58 HttpxSendArgs,
59 AsyncTransport,
60 RequestOptions,
61 HttpxRequestFiles,
62 ModelBuilderProtocol,
63)
64from ._utils import is_dict, is_list, asyncify, is_given, lru_cache, is_mapping
65from ._compat import model_copy, model_dump
66from ._models import GenericModel, FinalRequestOptions, validate_type, construct_type
67from ._response import (
68 APIResponse,
69 BaseAPIResponse,
70 AsyncAPIResponse,
71 extract_response_type,
72)
73from ._constants import (
74 DEFAULT_TIMEOUT,
75 MAX_RETRY_DELAY,
76 DEFAULT_MAX_RETRIES,
77 INITIAL_RETRY_DELAY,
78 RAW_RESPONSE_HEADER,
79 OVERRIDE_CAST_TO_HEADER,
80 DEFAULT_CONNECTION_LIMITS,
81)
82from ._streaming import Stream, SSEDecoder, AsyncStream, SSEBytesDecoder
83from ._exceptions import (
84 APIStatusError,
85 APITimeoutError,
86 APIConnectionError,
87 APIResponseValidationError,
88)
89from ._legacy_response import LegacyAPIResponse
90
91log: logging.Logger = logging.getLogger(__name__)
92
93# TODO: make base page type vars covariant
94SyncPageT = TypeVar("SyncPageT", bound="BaseSyncPage[Any]")
95AsyncPageT = TypeVar("AsyncPageT", bound="BaseAsyncPage[Any]")
96
97
98_T = TypeVar("_T")
99_T_co = TypeVar("_T_co", covariant=True)
100
101_StreamT = TypeVar("_StreamT", bound=Stream[Any])
102_AsyncStreamT = TypeVar("_AsyncStreamT", bound=AsyncStream[Any])
103
104if TYPE_CHECKING:
105 from httpx._config import DEFAULT_TIMEOUT_CONFIG as HTTPX_DEFAULT_TIMEOUT
106else:
107 try:
108 from httpx._config import DEFAULT_TIMEOUT_CONFIG as HTTPX_DEFAULT_TIMEOUT
109 except ImportError:
110 # taken from https://github.com/encode/httpx/blob/3ba5fe0d7ac70222590e759c31442b1cab263791/httpx/_config.py#L366
111 HTTPX_DEFAULT_TIMEOUT = Timeout(5.0)
112
113
114class PageInfo:
115 """Stores the necessary information to build the request to retrieve the next page.
116
117 Either `url` or `params` must be set.
118 """
119
120 url: URL | NotGiven
121 params: Query | NotGiven
122
123 @overload
124 def __init__(
125 self,
126 *,
127 url: URL,
128 ) -> None: ...
129
130 @overload
131 def __init__(
132 self,
133 *,
134 params: Query,
135 ) -> None: ...
136
137 def __init__(
138 self,
139 *,
140 url: URL | NotGiven = NOT_GIVEN,
141 params: Query | NotGiven = NOT_GIVEN,
142 ) -> None:
143 self.url = url
144 self.params = params
145
146
147class BasePage(GenericModel, Generic[_T]):
148 """
149 Defines the core interface for pagination.
150
151 Type Args:
152 ModelT: The pydantic model that represents an item in the response.
153
154 Methods:
155 has_next_page(): Check if there is another page available
156 next_page_info(): Get the necessary information to make a request for the next page
157 """
158
159 _options: FinalRequestOptions = PrivateAttr()
160 _model: Type[_T] = PrivateAttr()
161
162 def has_next_page(self) -> bool:
163 items = self._get_page_items()
164 if not items:
165 return False
166 return self.next_page_info() is not None
167
168 def next_page_info(self) -> Optional[PageInfo]: ...
169
170 def _get_page_items(self) -> Iterable[_T]: # type: ignore[empty-body]
171 ...
172
173 def _params_from_url(self, url: URL) -> httpx.QueryParams:
174 # TODO: do we have to preprocess params here?
175 return httpx.QueryParams(cast(Any, self._options.params)).merge(url.params)
176
177 def _info_to_options(self, info: PageInfo) -> FinalRequestOptions:
178 options = model_copy(self._options)
179 options._strip_raw_response_header()
180
181 if not isinstance(info.params, NotGiven):
182 options.params = {**options.params, **info.params}
183 return options
184
185 if not isinstance(info.url, NotGiven):
186 params = self._params_from_url(info.url)
187 url = info.url.copy_with(params=params)
188 options.params = dict(url.params)
189 options.url = str(url)
190 return options
191
192 raise ValueError("Unexpected PageInfo state")
193
194
195class BaseSyncPage(BasePage[_T], Generic[_T]):
196 _client: SyncAPIClient = pydantic.PrivateAttr()
197
198 def _set_private_attributes(
199 self,
200 client: SyncAPIClient,
201 model: Type[_T],
202 options: FinalRequestOptions,
203 ) -> None:
204 self._model = model
205 self._client = client
206 self._options = options
207
208 # Pydantic uses a custom `__iter__` method to support casting BaseModels
209 # to dictionaries. e.g. dict(model).
210 # As we want to support `for item in page`, this is inherently incompatible
211 # with the default pydantic behaviour. It is not possible to support both
212 # use cases at once. Fortunately, this is not a big deal as all other pydantic
213 # methods should continue to work as expected as there is an alternative method
214 # to cast a model to a dictionary, model.dict(), which is used internally
215 # by pydantic.
216 def __iter__(self) -> Iterator[_T]: # type: ignore
217 for page in self.iter_pages():
218 for item in page._get_page_items():
219 yield item
220
221 def iter_pages(self: SyncPageT) -> Iterator[SyncPageT]:
222 page = self
223 while True:
224 yield page
225 if page.has_next_page():
226 page = page.get_next_page()
227 else:
228 return
229
230 def get_next_page(self: SyncPageT) -> SyncPageT:
231 info = self.next_page_info()
232 if not info:
233 raise RuntimeError(
234 "No next page expected; please check `.has_next_page()` before calling `.get_next_page()`."
235 )
236
237 options = self._info_to_options(info)
238 return self._client._request_api_list(self._model, page=self.__class__, options=options)
239
240
241class AsyncPaginator(Generic[_T, AsyncPageT]):
242 def __init__(
243 self,
244 client: AsyncAPIClient,
245 options: FinalRequestOptions,
246 page_cls: Type[AsyncPageT],
247 model: Type[_T],
248 ) -> None:
249 self._model = model
250 self._client = client
251 self._options = options
252 self._page_cls = page_cls
253
254 def __await__(self) -> Generator[Any, None, AsyncPageT]:
255 return self._get_page().__await__()
256
257 async def _get_page(self) -> AsyncPageT:
258 def _parser(resp: AsyncPageT) -> AsyncPageT:
259 resp._set_private_attributes(
260 model=self._model,
261 options=self._options,
262 client=self._client,
263 )
264 return resp
265
266 self._options.post_parser = _parser
267
268 return await self._client.request(self._page_cls, self._options)
269
270 async def __aiter__(self) -> AsyncIterator[_T]:
271 # https://github.com/microsoft/pyright/issues/3464
272 page = cast(
273 AsyncPageT,
274 await self, # type: ignore
275 )
276 async for item in page:
277 yield item
278
279
280class BaseAsyncPage(BasePage[_T], Generic[_T]):
281 _client: AsyncAPIClient = pydantic.PrivateAttr()
282
283 def _set_private_attributes(
284 self,
285 model: Type[_T],
286 client: AsyncAPIClient,
287 options: FinalRequestOptions,
288 ) -> None:
289 self._model = model
290 self._client = client
291 self._options = options
292
293 async def __aiter__(self) -> AsyncIterator[_T]:
294 async for page in self.iter_pages():
295 for item in page._get_page_items():
296 yield item
297
298 async def iter_pages(self: AsyncPageT) -> AsyncIterator[AsyncPageT]:
299 page = self
300 while True:
301 yield page
302 if page.has_next_page():
303 page = await page.get_next_page()
304 else:
305 return
306
307 async def get_next_page(self: AsyncPageT) -> AsyncPageT:
308 info = self.next_page_info()
309 if not info:
310 raise RuntimeError(
311 "No next page expected; please check `.has_next_page()` before calling `.get_next_page()`."
312 )
313
314 options = self._info_to_options(info)
315 return await self._client._request_api_list(self._model, page=self.__class__, options=options)
316
317
318_HttpxClientT = TypeVar("_HttpxClientT", bound=Union[httpx.Client, httpx.AsyncClient])
319_DefaultStreamT = TypeVar("_DefaultStreamT", bound=Union[Stream[Any], AsyncStream[Any]])
320
321
322class BaseClient(Generic[_HttpxClientT, _DefaultStreamT]):
323 _client: _HttpxClientT
324 _version: str
325 _base_url: URL
326 max_retries: int
327 timeout: Union[float, Timeout, None]
328 _limits: httpx.Limits
329 _proxies: ProxiesTypes | None
330 _transport: Transport | AsyncTransport | None
331 _strict_response_validation: bool
332 _idempotency_header: str | None
333 _default_stream_cls: type[_DefaultStreamT] | None = None
334
335 def __init__(
336 self,
337 *,
338 version: str,
339 base_url: str | URL,
340 _strict_response_validation: bool,
341 max_retries: int = DEFAULT_MAX_RETRIES,
342 timeout: float | Timeout | None = DEFAULT_TIMEOUT,
343 limits: httpx.Limits,
344 transport: Transport | AsyncTransport | None,
345 proxies: ProxiesTypes | None,
346 custom_headers: Mapping[str, str] | None = None,
347 custom_query: Mapping[str, object] | None = None,
348 ) -> None:
349 self._version = version
350 self._base_url = self._enforce_trailing_slash(URL(base_url))
351 self.max_retries = max_retries
352 self.timeout = timeout
353 self._limits = limits
354 self._proxies = proxies
355 self._transport = transport
356 self._custom_headers = custom_headers or {}
357 self._custom_query = custom_query or {}
358 self._strict_response_validation = _strict_response_validation
359 self._idempotency_header = None
360 self._platform: Platform | None = None
361
362 if max_retries is None: # pyright: ignore[reportUnnecessaryComparison]
363 raise TypeError(
364 "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 `openai.DEFAULT_MAX_RETRIES`"
365 )
366
367 def _enforce_trailing_slash(self, url: URL) -> URL:
368 if url.raw_path.endswith(b"/"):
369 return url
370 return url.copy_with(raw_path=url.raw_path + b"/")
371
372 def _make_status_error_from_response(
373 self,
374 response: httpx.Response,
375 ) -> APIStatusError:
376 if response.is_closed and not response.is_stream_consumed:
377 # We can't read the response body as it has been closed
378 # before it was read. This can happen if an event hook
379 # raises a status error.
380 body = None
381 err_msg = f"Error code: {response.status_code}"
382 else:
383 err_text = response.text.strip()
384 body = err_text
385
386 try:
387 body = json.loads(err_text)
388 err_msg = f"Error code: {response.status_code} - {body}"
389 except Exception:
390 err_msg = err_text or f"Error code: {response.status_code}"
391
392 return self._make_status_error(err_msg, body=body, response=response)
393
394 def _make_status_error(
395 self,
396 err_msg: str,
397 *,
398 body: object,
399 response: httpx.Response,
400 ) -> _exceptions.APIStatusError:
401 raise NotImplementedError()
402
403 def _remaining_retries(
404 self,
405 remaining_retries: Optional[int],
406 options: FinalRequestOptions,
407 ) -> int:
408 return remaining_retries if remaining_retries is not None else options.get_max_retries(self.max_retries)
409
410 def _build_headers(self, options: FinalRequestOptions) -> httpx.Headers:
411 custom_headers = options.headers or {}
412 headers_dict = _merge_mappings(self.default_headers, custom_headers)
413 self._validate_headers(headers_dict, custom_headers)
414
415 # headers are case-insensitive while dictionaries are not.
416 headers = httpx.Headers(headers_dict)
417
418 idempotency_header = self._idempotency_header
419 if idempotency_header and options.method.lower() != "get" and idempotency_header not in headers:
420 headers[idempotency_header] = options.idempotency_key or self._idempotency_key()
421
422 return headers
423
424 def _prepare_url(self, url: str) -> URL:
425 """
426 Merge a URL argument together with any 'base_url' on the client,
427 to create the URL used for the outgoing request.
428 """
429 # Copied from httpx's `_merge_url` method.
430 merge_url = URL(url)
431 if merge_url.is_relative_url:
432 merge_raw_path = self.base_url.raw_path + merge_url.raw_path.lstrip(b"/")
433 return self.base_url.copy_with(raw_path=merge_raw_path)
434
435 return merge_url
436
437 def _make_sse_decoder(self) -> SSEDecoder | SSEBytesDecoder:
438 return SSEDecoder()
439
440 def _build_request(
441 self,
442 options: FinalRequestOptions,
443 ) -> httpx.Request:
444 if log.isEnabledFor(logging.DEBUG):
445 log.debug("Request options: %s", model_dump(options, exclude_unset=True))
446
447 kwargs: dict[str, Any] = {}
448
449 json_data = options.json_data
450 if options.extra_json is not None:
451 if json_data is None:
452 json_data = cast(Body, options.extra_json)
453 elif is_mapping(json_data):
454 json_data = _merge_mappings(json_data, options.extra_json)
455 else:
456 raise RuntimeError(f"Unexpected JSON data type, {type(json_data)}, cannot merge with `extra_body`")
457
458 headers = self._build_headers(options)
459 params = _merge_mappings(self.default_query, options.params)
460 content_type = headers.get("Content-Type")
461 files = options.files
462
463 # If the given Content-Type header is multipart/form-data then it
464 # has to be removed so that httpx can generate the header with
465 # additional information for us as it has to be in this form
466 # for the server to be able to correctly parse the request:
467 # multipart/form-data; boundary=---abc--
468 if content_type is not None and content_type.startswith("multipart/form-data"):
469 if "boundary" not in content_type:
470 # only remove the header if the boundary hasn't been explicitly set
471 # as the caller doesn't want httpx to come up with their own boundary
472 headers.pop("Content-Type")
473
474 # As we are now sending multipart/form-data instead of application/json
475 # we need to tell httpx to use it, https://www.python-httpx.org/advanced/clients/#multipart-file-encoding
476 if json_data:
477 if not is_dict(json_data):
478 raise TypeError(
479 f"Expected query input to be a dictionary for multipart requests but got {type(json_data)} instead."
480 )
481 kwargs["data"] = self._serialize_multipartform(json_data)
482
483 # httpx determines whether or not to send a "multipart/form-data"
484 # request based on the truthiness of the "files" argument.
485 # This gets around that issue by generating a dict value that
486 # evaluates to true.
487 #
488 # https://github.com/encode/httpx/discussions/2399#discussioncomment-3814186
489 if not files:
490 files = cast(HttpxRequestFiles, ForceMultipartDict())
491
492 # TODO: report this error to httpx
493 return self._client.build_request( # pyright: ignore[reportUnknownMemberType]
494 headers=headers,
495 timeout=self.timeout if isinstance(options.timeout, NotGiven) else options.timeout,
496 method=options.method,
497 url=self._prepare_url(options.url),
498 # the `Query` type that we use is incompatible with qs'
499 # `Params` type as it needs to be typed as `Mapping[str, object]`
500 # so that passing a `TypedDict` doesn't cause an error.
501 # https://github.com/microsoft/pyright/issues/3526#event-6715453066
502 params=self.qs.stringify(cast(Mapping[str, Any], params)) if params else None,
503 json=json_data,
504 files=files,
505 **kwargs,
506 )
507
508 def _serialize_multipartform(self, data: Mapping[object, object]) -> dict[str, object]:
509 items = self.qs.stringify_items(
510 # TODO: type ignore is required as stringify_items is well typed but we can't be
511 # well typed without heavy validation.
512 data, # type: ignore
513 array_format="brackets",
514 )
515 serialized: dict[str, object] = {}
516 for key, value in items:
517 existing = serialized.get(key)
518
519 if not existing:
520 serialized[key] = value
521 continue
522
523 # If a value has already been set for this key then that
524 # means we're sending data like `array[]=[1, 2, 3]` and we
525 # need to tell httpx that we want to send multiple values with
526 # the same key which is done by using a list or a tuple.
527 #
528 # Note: 2d arrays should never result in the same key at both
529 # levels so it's safe to assume that if the value is a list,
530 # it was because we changed it to be a list.
531 if is_list(existing):
532 existing.append(value)
533 else:
534 serialized[key] = [existing, value]
535
536 return serialized
537
538 def _maybe_override_cast_to(self, cast_to: type[ResponseT], options: FinalRequestOptions) -> type[ResponseT]:
539 if not is_given(options.headers):
540 return cast_to
541
542 # make a copy of the headers so we don't mutate user-input
543 headers = dict(options.headers)
544
545 # we internally support defining a temporary header to override the
546 # default `cast_to` type for use with `.with_raw_response` and `.with_streaming_response`
547 # see _response.py for implementation details
548 override_cast_to = headers.pop(OVERRIDE_CAST_TO_HEADER, NOT_GIVEN)
549 if is_given(override_cast_to):
550 options.headers = headers
551 return cast(Type[ResponseT], override_cast_to)
552
553 return cast_to
554
555 def _should_stream_response_body(self, request: httpx.Request) -> bool:
556 return request.headers.get(RAW_RESPONSE_HEADER) == "stream" # type: ignore[no-any-return]
557
558 def _process_response_data(
559 self,
560 *,
561 data: object,
562 cast_to: type[ResponseT],
563 response: httpx.Response,
564 ) -> ResponseT:
565 if data is None:
566 return cast(ResponseT, None)
567
568 if cast_to is object:
569 return cast(ResponseT, data)
570
571 try:
572 if inspect.isclass(cast_to) and issubclass(cast_to, ModelBuilderProtocol):
573 return cast(ResponseT, cast_to.build(response=response, data=data))
574
575 if self._strict_response_validation:
576 return cast(ResponseT, validate_type(type_=cast_to, value=data))
577
578 return cast(ResponseT, construct_type(type_=cast_to, value=data))
579 except pydantic.ValidationError as err:
580 raise APIResponseValidationError(response=response, body=data) from err
581
582 @property
583 def qs(self) -> Querystring:
584 return Querystring()
585
586 @property
587 def custom_auth(self) -> httpx.Auth | None:
588 return None
589
590 @property
591 def auth_headers(self) -> dict[str, str]:
592 return {}
593
594 @property
595 def default_headers(self) -> dict[str, str | Omit]:
596 return {
597 "Accept": "application/json",
598 "Content-Type": "application/json",
599 "User-Agent": self.user_agent,
600 **self.platform_headers(),
601 **self.auth_headers,
602 **self._custom_headers,
603 }
604
605 @property
606 def default_query(self) -> dict[str, object]:
607 return {
608 **self._custom_query,
609 }
610
611 def _validate_headers(
612 self,
613 headers: Headers, # noqa: ARG002
614 custom_headers: Headers, # noqa: ARG002
615 ) -> None:
616 """Validate the given default headers and custom headers.
617
618 Does nothing by default.
619 """
620 return
621
622 @property
623 def user_agent(self) -> str:
624 return f"{self.__class__.__name__}/Python {self._version}"
625
626 @property
627 def base_url(self) -> URL:
628 return self._base_url
629
630 @base_url.setter
631 def base_url(self, url: URL | str) -> None:
632 self._base_url = self._enforce_trailing_slash(url if isinstance(url, URL) else URL(url))
633
634 def platform_headers(self) -> Dict[str, str]:
635 # the actual implementation is in a separate `lru_cache` decorated
636 # function because adding `lru_cache` to methods will leak memory
637 # https://github.com/python/cpython/issues/88476
638 return platform_headers(self._version, platform=self._platform)
639
640 def _parse_retry_after_header(self, response_headers: Optional[httpx.Headers] = None) -> float | None:
641 """Returns a float of the number of seconds (not milliseconds) to wait after retrying, or None if unspecified.
642
643 About the Retry-After header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After
644 See also https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After#syntax
645 """
646 if response_headers is None:
647 return None
648
649 # First, try the non-standard `retry-after-ms` header for milliseconds,
650 # which is more precise than integer-seconds `retry-after`
651 try:
652 retry_ms_header = response_headers.get("retry-after-ms", None)
653 return float(retry_ms_header) / 1000
654 except (TypeError, ValueError):
655 pass
656
657 # Next, try parsing `retry-after` header as seconds (allowing nonstandard floats).
658 retry_header = response_headers.get("retry-after")
659 try:
660 # note: the spec indicates that this should only ever be an integer
661 # but if someone sends a float there's no reason for us to not respect it
662 return float(retry_header)
663 except (TypeError, ValueError):
664 pass
665
666 # Last, try parsing `retry-after` as a date.
667 retry_date_tuple = email.utils.parsedate_tz(retry_header)
668 if retry_date_tuple is None:
669 return None
670
671 retry_date = email.utils.mktime_tz(retry_date_tuple)
672 return float(retry_date - time.time())
673
674 def _calculate_retry_timeout(
675 self,
676 remaining_retries: int,
677 options: FinalRequestOptions,
678 response_headers: Optional[httpx.Headers] = None,
679 ) -> float:
680 max_retries = options.get_max_retries(self.max_retries)
681
682 # If the API asks us to wait a certain amount of time (and it's a reasonable amount), just do what it says.
683 retry_after = self._parse_retry_after_header(response_headers)
684 if retry_after is not None and 0 < retry_after <= 60:
685 return retry_after
686
687 nb_retries = max_retries - remaining_retries
688
689 # Apply exponential backoff, but not more than the max.
690 sleep_seconds = min(INITIAL_RETRY_DELAY * pow(2.0, nb_retries), MAX_RETRY_DELAY)
691
692 # Apply some jitter, plus-or-minus half a second.
693 jitter = 1 - 0.25 * random()
694 timeout = sleep_seconds * jitter
695 return timeout if timeout >= 0 else 0
696
697 def _should_retry(self, response: httpx.Response) -> bool:
698 # Note: this is not a standard header
699 should_retry_header = response.headers.get("x-should-retry")
700
701 # If the server explicitly says whether or not to retry, obey.
702 if should_retry_header == "true":
703 log.debug("Retrying as header `x-should-retry` is set to `true`")
704 return True
705 if should_retry_header == "false":
706 log.debug("Not retrying as header `x-should-retry` is set to `false`")
707 return False
708
709 # Retry on request timeouts.
710 if response.status_code == 408:
711 log.debug("Retrying due to status code %i", response.status_code)
712 return True
713
714 # Retry on lock timeouts.
715 if response.status_code == 409:
716 log.debug("Retrying due to status code %i", response.status_code)
717 return True
718
719 # Retry on rate limits.
720 if response.status_code == 429:
721 log.debug("Retrying due to status code %i", response.status_code)
722 return True
723
724 # Retry internal errors.
725 if response.status_code >= 500:
726 log.debug("Retrying due to status code %i", response.status_code)
727 return True
728
729 log.debug("Not retrying")
730 return False
731
732 def _idempotency_key(self) -> str:
733 return f"stainless-python-retry-{uuid.uuid4()}"
734
735
736class _DefaultHttpxClient(httpx.Client):
737 def __init__(self, **kwargs: Any) -> None:
738 kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
739 kwargs.setdefault("limits", DEFAULT_CONNECTION_LIMITS)
740 kwargs.setdefault("follow_redirects", True)
741 super().__init__(**kwargs)
742
743
744if TYPE_CHECKING:
745 DefaultHttpxClient = httpx.Client
746 """An alias to `httpx.Client` that provides the same defaults that this SDK
747 uses internally.
748
749 This is useful because overriding the `http_client` with your own instance of
750 `httpx.Client` will result in httpx's defaults being used, not ours.
751 """
752else:
753 DefaultHttpxClient = _DefaultHttpxClient
754
755
756class SyncHttpxClientWrapper(DefaultHttpxClient):
757 def __del__(self) -> None:
758 try:
759 self.close()
760 except Exception:
761 pass
762
763
764class SyncAPIClient(BaseClient[httpx.Client, Stream[Any]]):
765 _client: httpx.Client
766 _default_stream_cls: type[Stream[Any]] | None = None
767
768 def __init__(
769 self,
770 *,
771 version: str,
772 base_url: str | URL,
773 max_retries: int = DEFAULT_MAX_RETRIES,
774 timeout: float | Timeout | None | NotGiven = NOT_GIVEN,
775 transport: Transport | None = None,
776 proxies: ProxiesTypes | None = None,
777 limits: Limits | None = None,
778 http_client: httpx.Client | None = None,
779 custom_headers: Mapping[str, str] | None = None,
780 custom_query: Mapping[str, object] | None = None,
781 _strict_response_validation: bool,
782 ) -> None:
783 if limits is not None:
784 warnings.warn(
785 "The `connection_pool_limits` argument is deprecated. The `http_client` argument should be passed instead",
786 category=DeprecationWarning,
787 stacklevel=3,
788 )
789 if http_client is not None:
790 raise ValueError("The `http_client` argument is mutually exclusive with `connection_pool_limits`")
791 else:
792 limits = DEFAULT_CONNECTION_LIMITS
793
794 if transport is not None:
795 warnings.warn(
796 "The `transport` argument is deprecated. The `http_client` argument should be passed instead",
797 category=DeprecationWarning,
798 stacklevel=3,
799 )
800 if http_client is not None:
801 raise ValueError("The `http_client` argument is mutually exclusive with `transport`")
802
803 if proxies is not None:
804 warnings.warn(
805 "The `proxies` argument is deprecated. The `http_client` argument should be passed instead",
806 category=DeprecationWarning,
807 stacklevel=3,
808 )
809 if http_client is not None:
810 raise ValueError("The `http_client` argument is mutually exclusive with `proxies`")
811
812 if not is_given(timeout):
813 # if the user passed in a custom http client with a non-default
814 # timeout set then we use that timeout.
815 #
816 # note: there is an edge case here where the user passes in a client
817 # where they've explicitly set the timeout to match the default timeout
818 # as this check is structural, meaning that we'll think they didn't
819 # pass in a timeout and will ignore it
820 if http_client and http_client.timeout != HTTPX_DEFAULT_TIMEOUT:
821 timeout = http_client.timeout
822 else:
823 timeout = DEFAULT_TIMEOUT
824
825 if http_client is not None and not isinstance(http_client, httpx.Client): # pyright: ignore[reportUnnecessaryIsInstance]
826 raise TypeError(
827 f"Invalid `http_client` argument; Expected an instance of `httpx.Client` but got {type(http_client)}"
828 )
829
830 super().__init__(
831 version=version,
832 limits=limits,
833 # cast to a valid type because mypy doesn't understand our type narrowing
834 timeout=cast(Timeout, timeout),
835 proxies=proxies,
836 base_url=base_url,
837 transport=transport,
838 max_retries=max_retries,
839 custom_query=custom_query,
840 custom_headers=custom_headers,
841 _strict_response_validation=_strict_response_validation,
842 )
843 self._client = http_client or SyncHttpxClientWrapper(
844 base_url=base_url,
845 # cast to a valid type because mypy doesn't understand our type narrowing
846 timeout=cast(Timeout, timeout),
847 proxies=proxies,
848 transport=transport,
849 limits=limits,
850 follow_redirects=True,
851 )
852
853 def is_closed(self) -> bool:
854 return self._client.is_closed
855
856 def close(self) -> None:
857 """Close the underlying HTTPX client.
858
859 The client will *not* be usable after this.
860 """
861 # If an error is thrown while constructing a client, self._client
862 # may not be present
863 if hasattr(self, "_client"):
864 self._client.close()
865
866 def __enter__(self: _T) -> _T:
867 return self
868
869 def __exit__(
870 self,
871 exc_type: type[BaseException] | None,
872 exc: BaseException | None,
873 exc_tb: TracebackType | None,
874 ) -> None:
875 self.close()
876
877 def _prepare_options(
878 self,
879 options: FinalRequestOptions, # noqa: ARG002
880 ) -> FinalRequestOptions:
881 """Hook for mutating the given options"""
882 return options
883
884 def _prepare_request(
885 self,
886 request: httpx.Request, # noqa: ARG002
887 ) -> None:
888 """This method is used as a callback for mutating the `Request` object
889 after it has been constructed.
890 This is useful for cases where you want to add certain headers based off of
891 the request properties, e.g. `url`, `method` etc.
892 """
893 return None
894
895 @overload
896 def request(
897 self,
898 cast_to: Type[ResponseT],
899 options: FinalRequestOptions,
900 remaining_retries: Optional[int] = None,
901 *,
902 stream: Literal[True],
903 stream_cls: Type[_StreamT],
904 ) -> _StreamT: ...
905
906 @overload
907 def request(
908 self,
909 cast_to: Type[ResponseT],
910 options: FinalRequestOptions,
911 remaining_retries: Optional[int] = None,
912 *,
913 stream: Literal[False] = False,
914 ) -> ResponseT: ...
915
916 @overload
917 def request(
918 self,
919 cast_to: Type[ResponseT],
920 options: FinalRequestOptions,
921 remaining_retries: Optional[int] = None,
922 *,
923 stream: bool = False,
924 stream_cls: Type[_StreamT] | None = None,
925 ) -> ResponseT | _StreamT: ...
926
927 def request(
928 self,
929 cast_to: Type[ResponseT],
930 options: FinalRequestOptions,
931 remaining_retries: Optional[int] = None,
932 *,
933 stream: bool = False,
934 stream_cls: type[_StreamT] | None = None,
935 ) -> ResponseT | _StreamT:
936 return self._request(
937 cast_to=cast_to,
938 options=options,
939 stream=stream,
940 stream_cls=stream_cls,
941 remaining_retries=remaining_retries,
942 )
943
944 def _request(
945 self,
946 *,
947 cast_to: Type[ResponseT],
948 options: FinalRequestOptions,
949 remaining_retries: int | None,
950 stream: bool,
951 stream_cls: type[_StreamT] | None,
952 ) -> ResponseT | _StreamT:
953 # create a copy of the options we were given so that if the
954 # options are mutated later & we then retry, the retries are
955 # given the original options
956 input_options = model_copy(options)
957
958 cast_to = self._maybe_override_cast_to(cast_to, options)
959 options = self._prepare_options(options)
960
961 retries = self._remaining_retries(remaining_retries, options)
962 request = self._build_request(options)
963 self._prepare_request(request)
964
965 kwargs: HttpxSendArgs = {}
966 if self.custom_auth is not None:
967 kwargs["auth"] = self.custom_auth
968
969 log.debug("Sending HTTP Request: %s %s", request.method, request.url)
970
971 try:
972 response = self._client.send(
973 request,
974 stream=stream or self._should_stream_response_body(request=request),
975 **kwargs,
976 )
977 except httpx.TimeoutException as err:
978 log.debug("Encountered httpx.TimeoutException", exc_info=True)
979
980 if retries > 0:
981 return self._retry_request(
982 input_options,
983 cast_to,
984 retries,
985 stream=stream,
986 stream_cls=stream_cls,
987 response_headers=None,
988 )
989
990 log.debug("Raising timeout error")
991 raise APITimeoutError(request=request) from err
992 except Exception as err:
993 log.debug("Encountered Exception", exc_info=True)
994
995 if retries > 0:
996 return self._retry_request(
997 input_options,
998 cast_to,
999 retries,
1000 stream=stream,
1001 stream_cls=stream_cls,
1002 response_headers=None,
1003 )
1004
1005 log.debug("Raising connection error")
1006 raise APIConnectionError(request=request) from err
1007
1008 log.debug(
1009 'HTTP Response: %s %s "%i %s" %s',
1010 request.method,
1011 request.url,
1012 response.status_code,
1013 response.reason_phrase,
1014 response.headers,
1015 )
1016 log.debug("request_id: %s", response.headers.get("x-request-id"))
1017
1018 try:
1019 response.raise_for_status()
1020 except httpx.HTTPStatusError as err: # thrown on 4xx and 5xx status code
1021 log.debug("Encountered httpx.HTTPStatusError", exc_info=True)
1022
1023 if retries > 0 and self._should_retry(err.response):
1024 err.response.close()
1025 return self._retry_request(
1026 input_options,
1027 cast_to,
1028 retries,
1029 err.response.headers,
1030 stream=stream,
1031 stream_cls=stream_cls,
1032 )
1033
1034 # If the response is streamed then we need to explicitly read the response
1035 # to completion before attempting to access the response text.
1036 if not err.response.is_closed:
1037 err.response.read()
1038
1039 log.debug("Re-raising status error")
1040 raise self._make_status_error_from_response(err.response) from None
1041
1042 return self._process_response(
1043 cast_to=cast_to,
1044 options=options,
1045 response=response,
1046 stream=stream,
1047 stream_cls=stream_cls,
1048 retries_taken=options.get_max_retries(self.max_retries) - retries,
1049 )
1050
1051 def _retry_request(
1052 self,
1053 options: FinalRequestOptions,
1054 cast_to: Type[ResponseT],
1055 remaining_retries: int,
1056 response_headers: httpx.Headers | None,
1057 *,
1058 stream: bool,
1059 stream_cls: type[_StreamT] | None,
1060 ) -> ResponseT | _StreamT:
1061 remaining = remaining_retries - 1
1062 if remaining == 1:
1063 log.debug("1 retry left")
1064 else:
1065 log.debug("%i retries left", remaining)
1066
1067 timeout = self._calculate_retry_timeout(remaining, options, response_headers)
1068 log.info("Retrying request to %s in %f seconds", options.url, timeout)
1069
1070 # In a synchronous context we are blocking the entire thread. Up to the library user to run the client in a
1071 # different thread if necessary.
1072 time.sleep(timeout)
1073
1074 return self._request(
1075 options=options,
1076 cast_to=cast_to,
1077 remaining_retries=remaining,
1078 stream=stream,
1079 stream_cls=stream_cls,
1080 )
1081
1082 def _process_response(
1083 self,
1084 *,
1085 cast_to: Type[ResponseT],
1086 options: FinalRequestOptions,
1087 response: httpx.Response,
1088 stream: bool,
1089 stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None,
1090 retries_taken: int = 0,
1091 ) -> ResponseT:
1092 if response.request.headers.get(RAW_RESPONSE_HEADER) == "true":
1093 return cast(
1094 ResponseT,
1095 LegacyAPIResponse(
1096 raw=response,
1097 client=self,
1098 cast_to=cast_to,
1099 stream=stream,
1100 stream_cls=stream_cls,
1101 options=options,
1102 retries_taken=retries_taken,
1103 ),
1104 )
1105
1106 origin = get_origin(cast_to) or cast_to
1107
1108 if inspect.isclass(origin) and issubclass(origin, BaseAPIResponse):
1109 if not issubclass(origin, APIResponse):
1110 raise TypeError(f"API Response types must subclass {APIResponse}; Received {origin}")
1111
1112 response_cls = cast("type[BaseAPIResponse[Any]]", cast_to)
1113 return cast(
1114 ResponseT,
1115 response_cls(
1116 raw=response,
1117 client=self,
1118 cast_to=extract_response_type(response_cls),
1119 stream=stream,
1120 stream_cls=stream_cls,
1121 options=options,
1122 retries_taken=retries_taken,
1123 ),
1124 )
1125
1126 if cast_to == httpx.Response:
1127 return cast(ResponseT, response)
1128
1129 api_response = APIResponse(
1130 raw=response,
1131 client=self,
1132 cast_to=cast("type[ResponseT]", cast_to), # pyright: ignore[reportUnnecessaryCast]
1133 stream=stream,
1134 stream_cls=stream_cls,
1135 options=options,
1136 retries_taken=retries_taken,
1137 )
1138 if bool(response.request.headers.get(RAW_RESPONSE_HEADER)):
1139 return cast(ResponseT, api_response)
1140
1141 return api_response.parse()
1142
1143 def _request_api_list(
1144 self,
1145 model: Type[object],
1146 page: Type[SyncPageT],
1147 options: FinalRequestOptions,
1148 ) -> SyncPageT:
1149 def _parser(resp: SyncPageT) -> SyncPageT:
1150 resp._set_private_attributes(
1151 client=self,
1152 model=model,
1153 options=options,
1154 )
1155 return resp
1156
1157 options.post_parser = _parser
1158
1159 return self.request(page, options, stream=False)
1160
1161 @overload
1162 def get(
1163 self,
1164 path: str,
1165 *,
1166 cast_to: Type[ResponseT],
1167 options: RequestOptions = {},
1168 stream: Literal[False] = False,
1169 ) -> ResponseT: ...
1170
1171 @overload
1172 def get(
1173 self,
1174 path: str,
1175 *,
1176 cast_to: Type[ResponseT],
1177 options: RequestOptions = {},
1178 stream: Literal[True],
1179 stream_cls: type[_StreamT],
1180 ) -> _StreamT: ...
1181
1182 @overload
1183 def get(
1184 self,
1185 path: str,
1186 *,
1187 cast_to: Type[ResponseT],
1188 options: RequestOptions = {},
1189 stream: bool,
1190 stream_cls: type[_StreamT] | None = None,
1191 ) -> ResponseT | _StreamT: ...
1192
1193 def get(
1194 self,
1195 path: str,
1196 *,
1197 cast_to: Type[ResponseT],
1198 options: RequestOptions = {},
1199 stream: bool = False,
1200 stream_cls: type[_StreamT] | None = None,
1201 ) -> ResponseT | _StreamT:
1202 opts = FinalRequestOptions.construct(method="get", url=path, **options)
1203 # cast is required because mypy complains about returning Any even though
1204 # it understands the type variables
1205 return cast(ResponseT, self.request(cast_to, opts, stream=stream, stream_cls=stream_cls))
1206
1207 @overload
1208 def post(
1209 self,
1210 path: str,
1211 *,
1212 cast_to: Type[ResponseT],
1213 body: Body | None = None,
1214 options: RequestOptions = {},
1215 files: RequestFiles | None = None,
1216 stream: Literal[False] = False,
1217 ) -> ResponseT: ...
1218
1219 @overload
1220 def post(
1221 self,
1222 path: str,
1223 *,
1224 cast_to: Type[ResponseT],
1225 body: Body | None = None,
1226 options: RequestOptions = {},
1227 files: RequestFiles | None = None,
1228 stream: Literal[True],
1229 stream_cls: type[_StreamT],
1230 ) -> _StreamT: ...
1231
1232 @overload
1233 def post(
1234 self,
1235 path: str,
1236 *,
1237 cast_to: Type[ResponseT],
1238 body: Body | None = None,
1239 options: RequestOptions = {},
1240 files: RequestFiles | None = None,
1241 stream: bool,
1242 stream_cls: type[_StreamT] | None = None,
1243 ) -> ResponseT | _StreamT: ...
1244
1245 def post(
1246 self,
1247 path: str,
1248 *,
1249 cast_to: Type[ResponseT],
1250 body: Body | None = None,
1251 options: RequestOptions = {},
1252 files: RequestFiles | None = None,
1253 stream: bool = False,
1254 stream_cls: type[_StreamT] | None = None,
1255 ) -> ResponseT | _StreamT:
1256 opts = FinalRequestOptions.construct(
1257 method="post", url=path, json_data=body, files=to_httpx_files(files), **options
1258 )
1259 return cast(ResponseT, self.request(cast_to, opts, stream=stream, stream_cls=stream_cls))
1260
1261 def patch(
1262 self,
1263 path: str,
1264 *,
1265 cast_to: Type[ResponseT],
1266 body: Body | None = None,
1267 options: RequestOptions = {},
1268 ) -> ResponseT:
1269 opts = FinalRequestOptions.construct(method="patch", url=path, json_data=body, **options)
1270 return self.request(cast_to, opts)
1271
1272 def put(
1273 self,
1274 path: str,
1275 *,
1276 cast_to: Type[ResponseT],
1277 body: Body | None = None,
1278 files: RequestFiles | None = None,
1279 options: RequestOptions = {},
1280 ) -> ResponseT:
1281 opts = FinalRequestOptions.construct(
1282 method="put", url=path, json_data=body, files=to_httpx_files(files), **options
1283 )
1284 return self.request(cast_to, opts)
1285
1286 def delete(
1287 self,
1288 path: str,
1289 *,
1290 cast_to: Type[ResponseT],
1291 body: Body | None = None,
1292 options: RequestOptions = {},
1293 ) -> ResponseT:
1294 opts = FinalRequestOptions.construct(method="delete", url=path, json_data=body, **options)
1295 return self.request(cast_to, opts)
1296
1297 def get_api_list(
1298 self,
1299 path: str,
1300 *,
1301 model: Type[object],
1302 page: Type[SyncPageT],
1303 body: Body | None = None,
1304 options: RequestOptions = {},
1305 method: str = "get",
1306 ) -> SyncPageT:
1307 opts = FinalRequestOptions.construct(method=method, url=path, json_data=body, **options)
1308 return self._request_api_list(model, page, opts)
1309
1310
1311class _DefaultAsyncHttpxClient(httpx.AsyncClient):
1312 def __init__(self, **kwargs: Any) -> None:
1313 kwargs.setdefault("timeout", DEFAULT_TIMEOUT)
1314 kwargs.setdefault("limits", DEFAULT_CONNECTION_LIMITS)
1315 kwargs.setdefault("follow_redirects", True)
1316 super().__init__(**kwargs)
1317
1318
1319if TYPE_CHECKING:
1320 DefaultAsyncHttpxClient = httpx.AsyncClient
1321 """An alias to `httpx.AsyncClient` that provides the same defaults that this SDK
1322 uses internally.
1323
1324 This is useful because overriding the `http_client` with your own instance of
1325 `httpx.AsyncClient` will result in httpx's defaults being used, not ours.
1326 """
1327else:
1328 DefaultAsyncHttpxClient = _DefaultAsyncHttpxClient
1329
1330
1331class AsyncHttpxClientWrapper(DefaultAsyncHttpxClient):
1332 def __del__(self) -> None:
1333 try:
1334 # TODO(someday): support non asyncio runtimes here
1335 asyncio.get_running_loop().create_task(self.aclose())
1336 except Exception:
1337 pass
1338
1339
1340class AsyncAPIClient(BaseClient[httpx.AsyncClient, AsyncStream[Any]]):
1341 _client: httpx.AsyncClient
1342 _default_stream_cls: type[AsyncStream[Any]] | None = None
1343
1344 def __init__(
1345 self,
1346 *,
1347 version: str,
1348 base_url: str | URL,
1349 _strict_response_validation: bool,
1350 max_retries: int = DEFAULT_MAX_RETRIES,
1351 timeout: float | Timeout | None | NotGiven = NOT_GIVEN,
1352 transport: AsyncTransport | None = None,
1353 proxies: ProxiesTypes | None = None,
1354 limits: Limits | None = None,
1355 http_client: httpx.AsyncClient | None = None,
1356 custom_headers: Mapping[str, str] | None = None,
1357 custom_query: Mapping[str, object] | None = None,
1358 ) -> None:
1359 if limits is not None:
1360 warnings.warn(
1361 "The `connection_pool_limits` argument is deprecated. The `http_client` argument should be passed instead",
1362 category=DeprecationWarning,
1363 stacklevel=3,
1364 )
1365 if http_client is not None:
1366 raise ValueError("The `http_client` argument is mutually exclusive with `connection_pool_limits`")
1367 else:
1368 limits = DEFAULT_CONNECTION_LIMITS
1369
1370 if transport is not None:
1371 warnings.warn(
1372 "The `transport` argument is deprecated. The `http_client` argument should be passed instead",
1373 category=DeprecationWarning,
1374 stacklevel=3,
1375 )
1376 if http_client is not None:
1377 raise ValueError("The `http_client` argument is mutually exclusive with `transport`")
1378
1379 if proxies is not None:
1380 warnings.warn(
1381 "The `proxies` argument is deprecated. The `http_client` argument should be passed instead",
1382 category=DeprecationWarning,
1383 stacklevel=3,
1384 )
1385 if http_client is not None:
1386 raise ValueError("The `http_client` argument is mutually exclusive with `proxies`")
1387
1388 if not is_given(timeout):
1389 # if the user passed in a custom http client with a non-default
1390 # timeout set then we use that timeout.
1391 #
1392 # note: there is an edge case here where the user passes in a client
1393 # where they've explicitly set the timeout to match the default timeout
1394 # as this check is structural, meaning that we'll think they didn't
1395 # pass in a timeout and will ignore it
1396 if http_client and http_client.timeout != HTTPX_DEFAULT_TIMEOUT:
1397 timeout = http_client.timeout
1398 else:
1399 timeout = DEFAULT_TIMEOUT
1400
1401 if http_client is not None and not isinstance(http_client, httpx.AsyncClient): # pyright: ignore[reportUnnecessaryIsInstance]
1402 raise TypeError(
1403 f"Invalid `http_client` argument; Expected an instance of `httpx.AsyncClient` but got {type(http_client)}"
1404 )
1405
1406 super().__init__(
1407 version=version,
1408 base_url=base_url,
1409 limits=limits,
1410 # cast to a valid type because mypy doesn't understand our type narrowing
1411 timeout=cast(Timeout, timeout),
1412 proxies=proxies,
1413 transport=transport,
1414 max_retries=max_retries,
1415 custom_query=custom_query,
1416 custom_headers=custom_headers,
1417 _strict_response_validation=_strict_response_validation,
1418 )
1419 self._client = http_client or AsyncHttpxClientWrapper(
1420 base_url=base_url,
1421 # cast to a valid type because mypy doesn't understand our type narrowing
1422 timeout=cast(Timeout, timeout),
1423 proxies=proxies,
1424 transport=transport,
1425 limits=limits,
1426 follow_redirects=True,
1427 )
1428
1429 def is_closed(self) -> bool:
1430 return self._client.is_closed
1431
1432 async def close(self) -> None:
1433 """Close the underlying HTTPX client.
1434
1435 The client will *not* be usable after this.
1436 """
1437 await self._client.aclose()
1438
1439 async def __aenter__(self: _T) -> _T:
1440 return self
1441
1442 async def __aexit__(
1443 self,
1444 exc_type: type[BaseException] | None,
1445 exc: BaseException | None,
1446 exc_tb: TracebackType | None,
1447 ) -> None:
1448 await self.close()
1449
1450 async def _prepare_options(
1451 self,
1452 options: FinalRequestOptions, # noqa: ARG002
1453 ) -> FinalRequestOptions:
1454 """Hook for mutating the given options"""
1455 return options
1456
1457 async def _prepare_request(
1458 self,
1459 request: httpx.Request, # noqa: ARG002
1460 ) -> None:
1461 """This method is used as a callback for mutating the `Request` object
1462 after it has been constructed.
1463 This is useful for cases where you want to add certain headers based off of
1464 the request properties, e.g. `url`, `method` etc.
1465 """
1466 return None
1467
1468 @overload
1469 async def request(
1470 self,
1471 cast_to: Type[ResponseT],
1472 options: FinalRequestOptions,
1473 *,
1474 stream: Literal[False] = False,
1475 remaining_retries: Optional[int] = None,
1476 ) -> ResponseT: ...
1477
1478 @overload
1479 async def request(
1480 self,
1481 cast_to: Type[ResponseT],
1482 options: FinalRequestOptions,
1483 *,
1484 stream: Literal[True],
1485 stream_cls: type[_AsyncStreamT],
1486 remaining_retries: Optional[int] = None,
1487 ) -> _AsyncStreamT: ...
1488
1489 @overload
1490 async def request(
1491 self,
1492 cast_to: Type[ResponseT],
1493 options: FinalRequestOptions,
1494 *,
1495 stream: bool,
1496 stream_cls: type[_AsyncStreamT] | None = None,
1497 remaining_retries: Optional[int] = None,
1498 ) -> ResponseT | _AsyncStreamT: ...
1499
1500 async def request(
1501 self,
1502 cast_to: Type[ResponseT],
1503 options: FinalRequestOptions,
1504 *,
1505 stream: bool = False,
1506 stream_cls: type[_AsyncStreamT] | None = None,
1507 remaining_retries: Optional[int] = None,
1508 ) -> ResponseT | _AsyncStreamT:
1509 return await self._request(
1510 cast_to=cast_to,
1511 options=options,
1512 stream=stream,
1513 stream_cls=stream_cls,
1514 remaining_retries=remaining_retries,
1515 )
1516
1517 async def _request(
1518 self,
1519 cast_to: Type[ResponseT],
1520 options: FinalRequestOptions,
1521 *,
1522 stream: bool,
1523 stream_cls: type[_AsyncStreamT] | None,
1524 remaining_retries: int | None,
1525 ) -> ResponseT | _AsyncStreamT:
1526 if self._platform is None:
1527 # `get_platform` can make blocking IO calls so we
1528 # execute it earlier while we are in an async context
1529 self._platform = await asyncify(get_platform)()
1530
1531 # create a copy of the options we were given so that if the
1532 # options are mutated later & we then retry, the retries are
1533 # given the original options
1534 input_options = model_copy(options)
1535
1536 cast_to = self._maybe_override_cast_to(cast_to, options)
1537 options = await self._prepare_options(options)
1538
1539 retries = self._remaining_retries(remaining_retries, options)
1540 request = self._build_request(options)
1541 await self._prepare_request(request)
1542
1543 kwargs: HttpxSendArgs = {}
1544 if self.custom_auth is not None:
1545 kwargs["auth"] = self.custom_auth
1546
1547 try:
1548 response = await self._client.send(
1549 request,
1550 stream=stream or self._should_stream_response_body(request=request),
1551 **kwargs,
1552 )
1553 except httpx.TimeoutException as err:
1554 log.debug("Encountered httpx.TimeoutException", exc_info=True)
1555
1556 if retries > 0:
1557 return await self._retry_request(
1558 input_options,
1559 cast_to,
1560 retries,
1561 stream=stream,
1562 stream_cls=stream_cls,
1563 response_headers=None,
1564 )
1565
1566 log.debug("Raising timeout error")
1567 raise APITimeoutError(request=request) from err
1568 except Exception as err:
1569 log.debug("Encountered Exception", exc_info=True)
1570
1571 if retries > 0:
1572 return await self._retry_request(
1573 input_options,
1574 cast_to,
1575 retries,
1576 stream=stream,
1577 stream_cls=stream_cls,
1578 response_headers=None,
1579 )
1580
1581 log.debug("Raising connection error")
1582 raise APIConnectionError(request=request) from err
1583
1584 log.debug(
1585 'HTTP Request: %s %s "%i %s"', request.method, request.url, response.status_code, response.reason_phrase
1586 )
1587
1588 try:
1589 response.raise_for_status()
1590 except httpx.HTTPStatusError as err: # thrown on 4xx and 5xx status code
1591 log.debug("Encountered httpx.HTTPStatusError", exc_info=True)
1592
1593 if retries > 0 and self._should_retry(err.response):
1594 await err.response.aclose()
1595 return await self._retry_request(
1596 input_options,
1597 cast_to,
1598 retries,
1599 err.response.headers,
1600 stream=stream,
1601 stream_cls=stream_cls,
1602 )
1603
1604 # If the response is streamed then we need to explicitly read the response
1605 # to completion before attempting to access the response text.
1606 if not err.response.is_closed:
1607 await err.response.aread()
1608
1609 log.debug("Re-raising status error")
1610 raise self._make_status_error_from_response(err.response) from None
1611
1612 return await self._process_response(
1613 cast_to=cast_to,
1614 options=options,
1615 response=response,
1616 stream=stream,
1617 stream_cls=stream_cls,
1618 retries_taken=options.get_max_retries(self.max_retries) - retries,
1619 )
1620
1621 async def _retry_request(
1622 self,
1623 options: FinalRequestOptions,
1624 cast_to: Type[ResponseT],
1625 remaining_retries: int,
1626 response_headers: httpx.Headers | None,
1627 *,
1628 stream: bool,
1629 stream_cls: type[_AsyncStreamT] | None,
1630 ) -> ResponseT | _AsyncStreamT:
1631 remaining = remaining_retries - 1
1632 if remaining == 1:
1633 log.debug("1 retry left")
1634 else:
1635 log.debug("%i retries left", remaining)
1636
1637 timeout = self._calculate_retry_timeout(remaining, options, response_headers)
1638 log.info("Retrying request to %s in %f seconds", options.url, timeout)
1639
1640 await anyio.sleep(timeout)
1641
1642 return await self._request(
1643 options=options,
1644 cast_to=cast_to,
1645 remaining_retries=remaining,
1646 stream=stream,
1647 stream_cls=stream_cls,
1648 )
1649
1650 async def _process_response(
1651 self,
1652 *,
1653 cast_to: Type[ResponseT],
1654 options: FinalRequestOptions,
1655 response: httpx.Response,
1656 stream: bool,
1657 stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None,
1658 retries_taken: int = 0,
1659 ) -> ResponseT:
1660 if response.request.headers.get(RAW_RESPONSE_HEADER) == "true":
1661 return cast(
1662 ResponseT,
1663 LegacyAPIResponse(
1664 raw=response,
1665 client=self,
1666 cast_to=cast_to,
1667 stream=stream,
1668 stream_cls=stream_cls,
1669 options=options,
1670 retries_taken=retries_taken,
1671 ),
1672 )
1673
1674 origin = get_origin(cast_to) or cast_to
1675
1676 if inspect.isclass(origin) and issubclass(origin, BaseAPIResponse):
1677 if not issubclass(origin, AsyncAPIResponse):
1678 raise TypeError(f"API Response types must subclass {AsyncAPIResponse}; Received {origin}")
1679
1680 response_cls = cast("type[BaseAPIResponse[Any]]", cast_to)
1681 return cast(
1682 "ResponseT",
1683 response_cls(
1684 raw=response,
1685 client=self,
1686 cast_to=extract_response_type(response_cls),
1687 stream=stream,
1688 stream_cls=stream_cls,
1689 options=options,
1690 retries_taken=retries_taken,
1691 ),
1692 )
1693
1694 if cast_to == httpx.Response:
1695 return cast(ResponseT, response)
1696
1697 api_response = AsyncAPIResponse(
1698 raw=response,
1699 client=self,
1700 cast_to=cast("type[ResponseT]", cast_to), # pyright: ignore[reportUnnecessaryCast]
1701 stream=stream,
1702 stream_cls=stream_cls,
1703 options=options,
1704 retries_taken=retries_taken,
1705 )
1706 if bool(response.request.headers.get(RAW_RESPONSE_HEADER)):
1707 return cast(ResponseT, api_response)
1708
1709 return await api_response.parse()
1710
1711 def _request_api_list(
1712 self,
1713 model: Type[_T],
1714 page: Type[AsyncPageT],
1715 options: FinalRequestOptions,
1716 ) -> AsyncPaginator[_T, AsyncPageT]:
1717 return AsyncPaginator(client=self, options=options, page_cls=page, model=model)
1718
1719 @overload
1720 async def get(
1721 self,
1722 path: str,
1723 *,
1724 cast_to: Type[ResponseT],
1725 options: RequestOptions = {},
1726 stream: Literal[False] = False,
1727 ) -> ResponseT: ...
1728
1729 @overload
1730 async def get(
1731 self,
1732 path: str,
1733 *,
1734 cast_to: Type[ResponseT],
1735 options: RequestOptions = {},
1736 stream: Literal[True],
1737 stream_cls: type[_AsyncStreamT],
1738 ) -> _AsyncStreamT: ...
1739
1740 @overload
1741 async def get(
1742 self,
1743 path: str,
1744 *,
1745 cast_to: Type[ResponseT],
1746 options: RequestOptions = {},
1747 stream: bool,
1748 stream_cls: type[_AsyncStreamT] | None = None,
1749 ) -> ResponseT | _AsyncStreamT: ...
1750
1751 async def get(
1752 self,
1753 path: str,
1754 *,
1755 cast_to: Type[ResponseT],
1756 options: RequestOptions = {},
1757 stream: bool = False,
1758 stream_cls: type[_AsyncStreamT] | None = None,
1759 ) -> ResponseT | _AsyncStreamT:
1760 opts = FinalRequestOptions.construct(method="get", url=path, **options)
1761 return await self.request(cast_to, opts, stream=stream, stream_cls=stream_cls)
1762
1763 @overload
1764 async def post(
1765 self,
1766 path: str,
1767 *,
1768 cast_to: Type[ResponseT],
1769 body: Body | None = None,
1770 files: RequestFiles | None = None,
1771 options: RequestOptions = {},
1772 stream: Literal[False] = False,
1773 ) -> ResponseT: ...
1774
1775 @overload
1776 async def post(
1777 self,
1778 path: str,
1779 *,
1780 cast_to: Type[ResponseT],
1781 body: Body | None = None,
1782 files: RequestFiles | None = None,
1783 options: RequestOptions = {},
1784 stream: Literal[True],
1785 stream_cls: type[_AsyncStreamT],
1786 ) -> _AsyncStreamT: ...
1787
1788 @overload
1789 async def post(
1790 self,
1791 path: str,
1792 *,
1793 cast_to: Type[ResponseT],
1794 body: Body | None = None,
1795 files: RequestFiles | None = None,
1796 options: RequestOptions = {},
1797 stream: bool,
1798 stream_cls: type[_AsyncStreamT] | None = None,
1799 ) -> ResponseT | _AsyncStreamT: ...
1800
1801 async def post(
1802 self,
1803 path: str,
1804 *,
1805 cast_to: Type[ResponseT],
1806 body: Body | None = None,
1807 files: RequestFiles | None = None,
1808 options: RequestOptions = {},
1809 stream: bool = False,
1810 stream_cls: type[_AsyncStreamT] | None = None,
1811 ) -> ResponseT | _AsyncStreamT:
1812 opts = FinalRequestOptions.construct(
1813 method="post", url=path, json_data=body, files=await async_to_httpx_files(files), **options
1814 )
1815 return await self.request(cast_to, opts, stream=stream, stream_cls=stream_cls)
1816
1817 async def patch(
1818 self,
1819 path: str,
1820 *,
1821 cast_to: Type[ResponseT],
1822 body: Body | None = None,
1823 options: RequestOptions = {},
1824 ) -> ResponseT:
1825 opts = FinalRequestOptions.construct(method="patch", url=path, json_data=body, **options)
1826 return await self.request(cast_to, opts)
1827
1828 async def put(
1829 self,
1830 path: str,
1831 *,
1832 cast_to: Type[ResponseT],
1833 body: Body | None = None,
1834 files: RequestFiles | None = None,
1835 options: RequestOptions = {},
1836 ) -> ResponseT:
1837 opts = FinalRequestOptions.construct(
1838 method="put", url=path, json_data=body, files=await async_to_httpx_files(files), **options
1839 )
1840 return await self.request(cast_to, opts)
1841
1842 async def delete(
1843 self,
1844 path: str,
1845 *,
1846 cast_to: Type[ResponseT],
1847 body: Body | None = None,
1848 options: RequestOptions = {},
1849 ) -> ResponseT:
1850 opts = FinalRequestOptions.construct(method="delete", url=path, json_data=body, **options)
1851 return await self.request(cast_to, opts)
1852
1853 def get_api_list(
1854 self,
1855 path: str,
1856 *,
1857 model: Type[_T],
1858 page: Type[AsyncPageT],
1859 body: Body | None = None,
1860 options: RequestOptions = {},
1861 method: str = "get",
1862 ) -> AsyncPaginator[_T, AsyncPageT]:
1863 opts = FinalRequestOptions.construct(method=method, url=path, json_data=body, **options)
1864 return self._request_api_list(model, page, opts)
1865
1866
1867def make_request_options(
1868 *,
1869 query: Query | None = None,
1870 extra_headers: Headers | None = None,
1871 extra_query: Query | None = None,
1872 extra_body: Body | None = None,
1873 idempotency_key: str | None = None,
1874 timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
1875 post_parser: PostParser | NotGiven = NOT_GIVEN,
1876) -> RequestOptions:
1877 """Create a dict of type RequestOptions without keys of NotGiven values."""
1878 options: RequestOptions = {}
1879 if extra_headers is not None:
1880 options["headers"] = extra_headers
1881
1882 if extra_body is not None:
1883 options["extra_json"] = cast(AnyMapping, extra_body)
1884
1885 if query is not None:
1886 options["params"] = query
1887
1888 if extra_query is not None:
1889 options["params"] = {**options.get("params", {}), **extra_query}
1890
1891 if not isinstance(timeout, NotGiven):
1892 options["timeout"] = timeout
1893
1894 if idempotency_key is not None:
1895 options["idempotency_key"] = idempotency_key
1896
1897 if is_given(post_parser):
1898 # internal
1899 options["post_parser"] = post_parser # type: ignore
1900
1901 return options
1902
1903
1904class ForceMultipartDict(Dict[str, None]):
1905 def __bool__(self) -> bool:
1906 return True
1907
1908
1909class OtherPlatform:
1910 def __init__(self, name: str) -> None:
1911 self.name = name
1912
1913 @override
1914 def __str__(self) -> str:
1915 return f"Other:{self.name}"
1916
1917
1918Platform = Union[
1919 OtherPlatform,
1920 Literal[
1921 "MacOS",
1922 "Linux",
1923 "Windows",
1924 "FreeBSD",
1925 "OpenBSD",
1926 "iOS",
1927 "Android",
1928 "Unknown",
1929 ],
1930]
1931
1932
1933def get_platform() -> Platform:
1934 try:
1935 system = platform.system().lower()
1936 platform_name = platform.platform().lower()
1937 except Exception:
1938 return "Unknown"
1939
1940 if "iphone" in platform_name or "ipad" in platform_name:
1941 # Tested using Python3IDE on an iPhone 11 and Pythonista on an iPad 7
1942 # system is Darwin and platform_name is a string like:
1943 # - Darwin-21.6.0-iPhone12,1-64bit
1944 # - Darwin-21.6.0-iPad7,11-64bit
1945 return "iOS"
1946
1947 if system == "darwin":
1948 return "MacOS"
1949
1950 if system == "windows":
1951 return "Windows"
1952
1953 if "android" in platform_name:
1954 # Tested using Pydroid 3
1955 # system is Linux and platform_name is a string like 'Linux-5.10.81-android12-9-00001-geba40aecb3b7-ab8534902-aarch64-with-libc'
1956 return "Android"
1957
1958 if system == "linux":
1959 # https://distro.readthedocs.io/en/latest/#distro.id
1960 distro_id = distro.id()
1961 if distro_id == "freebsd":
1962 return "FreeBSD"
1963
1964 if distro_id == "openbsd":
1965 return "OpenBSD"
1966
1967 return "Linux"
1968
1969 if platform_name:
1970 return OtherPlatform(platform_name)
1971
1972 return "Unknown"
1973
1974
1975@lru_cache(maxsize=None)
1976def platform_headers(version: str, *, platform: Platform | None) -> Dict[str, str]:
1977 return {
1978 "X-Stainless-Lang": "python",
1979 "X-Stainless-Package-Version": version,
1980 "X-Stainless-OS": str(platform or get_platform()),
1981 "X-Stainless-Arch": str(get_architecture()),
1982 "X-Stainless-Runtime": get_python_runtime(),
1983 "X-Stainless-Runtime-Version": get_python_version(),
1984 }
1985
1986
1987class OtherArch:
1988 def __init__(self, name: str) -> None:
1989 self.name = name
1990
1991 @override
1992 def __str__(self) -> str:
1993 return f"other:{self.name}"
1994
1995
1996Arch = Union[OtherArch, Literal["x32", "x64", "arm", "arm64", "unknown"]]
1997
1998
1999def get_python_runtime() -> str:
2000 try:
2001 return platform.python_implementation()
2002 except Exception:
2003 return "unknown"
2004
2005
2006def get_python_version() -> str:
2007 try:
2008 return platform.python_version()
2009 except Exception:
2010 return "unknown"
2011
2012
2013def get_architecture() -> Arch:
2014 try:
2015 python_bitness, _ = platform.architecture()
2016 machine = platform.machine().lower()
2017 except Exception:
2018 return "unknown"
2019
2020 if machine in ("arm64", "aarch64"):
2021 return "arm64"
2022
2023 # TODO: untested
2024 if machine == "arm":
2025 return "arm"
2026
2027 if machine == "x86_64":
2028 return "x64"
2029
2030 # TODO: untested
2031 if python_bitness == "32bit":
2032 return "x32"
2033
2034 if machine:
2035 return OtherArch(machine)
2036
2037 return "unknown"
2038
2039
2040def _merge_mappings(
2041 obj1: Mapping[_T_co, Union[_T, Omit]],
2042 obj2: Mapping[_T_co, Union[_T, Omit]],
2043) -> Dict[_T_co, _T]:
2044 """Merge two mappings of the same type, removing any values that are instances of `Omit`.
2045
2046 In cases with duplicate keys the second mapping takes precedence.
2047 """
2048 merged = {**obj1, **obj2}
2049 return {key: value for key, value in merged.items() if not isinstance(value, Omit)}
2050