openai/openai-python
Publicmirrored from https://github.com/openai/openai-pythonAvailable
src/openai/_legacy_response.py
460lines · modecode
| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import inspect |
| 5 | import logging |
| 6 | import datetime |
| 7 | import functools |
| 8 | from typing import TYPE_CHECKING, Any, Union, Generic, TypeVar, Callable, Iterator, AsyncIterator, cast, overload |
| 9 | from typing_extensions import Awaitable, ParamSpec, override, deprecated, get_origin |
| 10 | |
| 11 | import anyio |
| 12 | import httpx |
| 13 | import pydantic |
| 14 | |
| 15 | from ._types import NoneType |
| 16 | from ._utils import is_given, extract_type_arg, is_annotated_type |
| 17 | from ._models import BaseModel, is_basemodel |
| 18 | from ._constants import RAW_RESPONSE_HEADER |
| 19 | from ._streaming import Stream, AsyncStream, is_stream_class_type, extract_stream_chunk_type |
| 20 | from ._exceptions import APIResponseValidationError |
| 21 | |
| 22 | if TYPE_CHECKING: |
| 23 | from ._models import FinalRequestOptions |
| 24 | from ._base_client import BaseClient |
| 25 | |
| 26 | |
| 27 | P = ParamSpec("P") |
| 28 | R = TypeVar("R") |
| 29 | _T = TypeVar("_T") |
| 30 | |
| 31 | log: logging.Logger = logging.getLogger(__name__) |
| 32 | |
| 33 | |
| 34 | class LegacyAPIResponse(Generic[R]): |
| 35 | """This is a legacy class as it will be replaced by `APIResponse` |
| 36 | and `AsyncAPIResponse` in the `_response.py` file in the next major |
| 37 | release. |
| 38 | |
| 39 | For the sync client this will mostly be the same with the exception |
| 40 | of `content` & `text` will be methods instead of properties. In the |
| 41 | async client, all methods will be async. |
| 42 | |
| 43 | A migration script will be provided & the migration in general should |
| 44 | be smooth. |
| 45 | """ |
| 46 | |
| 47 | _cast_to: type[R] |
| 48 | _client: BaseClient[Any, Any] |
| 49 | _parsed_by_type: dict[type[Any], Any] |
| 50 | _stream: bool |
| 51 | _stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None |
| 52 | _options: FinalRequestOptions |
| 53 | |
| 54 | http_response: httpx.Response |
| 55 | |
| 56 | def __init__( |
| 57 | self, |
| 58 | *, |
| 59 | raw: httpx.Response, |
| 60 | cast_to: type[R], |
| 61 | client: BaseClient[Any, Any], |
| 62 | stream: bool, |
| 63 | stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None, |
| 64 | options: FinalRequestOptions, |
| 65 | ) -> None: |
| 66 | self._cast_to = cast_to |
| 67 | self._client = client |
| 68 | self._parsed_by_type = {} |
| 69 | self._stream = stream |
| 70 | self._stream_cls = stream_cls |
| 71 | self._options = options |
| 72 | self.http_response = raw |
| 73 | |
| 74 | @property |
| 75 | def request_id(self) -> str | None: |
| 76 | return self.http_response.headers.get("x-request-id") # type: ignore[no-any-return] |
| 77 | |
| 78 | @overload |
| 79 | def parse(self, *, to: type[_T]) -> _T: |
| 80 | ... |
| 81 | |
| 82 | @overload |
| 83 | def parse(self) -> R: |
| 84 | ... |
| 85 | |
| 86 | def parse(self, *, to: type[_T] | None = None) -> R | _T: |
| 87 | """Returns the rich python representation of this response's data. |
| 88 | |
| 89 | NOTE: For the async client: this will become a coroutine in the next major version. |
| 90 | |
| 91 | For lower-level control, see `.read()`, `.json()`, `.iter_bytes()`. |
| 92 | |
| 93 | You can customise the type that the response is parsed into through |
| 94 | the `to` argument, e.g. |
| 95 | |
| 96 | ```py |
| 97 | from openai import BaseModel |
| 98 | |
| 99 | |
| 100 | class MyModel(BaseModel): |
| 101 | foo: str |
| 102 | |
| 103 | |
| 104 | obj = response.parse(to=MyModel) |
| 105 | print(obj.foo) |
| 106 | ``` |
| 107 | |
| 108 | We support parsing: |
| 109 | - `BaseModel` |
| 110 | - `dict` |
| 111 | - `list` |
| 112 | - `Union` |
| 113 | - `str` |
| 114 | - `int` |
| 115 | - `float` |
| 116 | - `httpx.Response` |
| 117 | """ |
| 118 | cache_key = to if to is not None else self._cast_to |
| 119 | cached = self._parsed_by_type.get(cache_key) |
| 120 | if cached is not None: |
| 121 | return cached # type: ignore[no-any-return] |
| 122 | |
| 123 | parsed = self._parse(to=to) |
| 124 | if is_given(self._options.post_parser): |
| 125 | parsed = self._options.post_parser(parsed) |
| 126 | |
| 127 | self._parsed_by_type[cache_key] = parsed |
| 128 | return parsed |
| 129 | |
| 130 | @property |
| 131 | def headers(self) -> httpx.Headers: |
| 132 | return self.http_response.headers |
| 133 | |
| 134 | @property |
| 135 | def http_request(self) -> httpx.Request: |
| 136 | return self.http_response.request |
| 137 | |
| 138 | @property |
| 139 | def status_code(self) -> int: |
| 140 | return self.http_response.status_code |
| 141 | |
| 142 | @property |
| 143 | def url(self) -> httpx.URL: |
| 144 | return self.http_response.url |
| 145 | |
| 146 | @property |
| 147 | def method(self) -> str: |
| 148 | return self.http_request.method |
| 149 | |
| 150 | @property |
| 151 | def content(self) -> bytes: |
| 152 | """Return the binary response content. |
| 153 | |
| 154 | NOTE: this will be removed in favour of `.read()` in the |
| 155 | next major version. |
| 156 | """ |
| 157 | return self.http_response.content |
| 158 | |
| 159 | @property |
| 160 | def text(self) -> str: |
| 161 | """Return the decoded response content. |
| 162 | |
| 163 | NOTE: this will be turned into a method in the next major version. |
| 164 | """ |
| 165 | return self.http_response.text |
| 166 | |
| 167 | @property |
| 168 | def http_version(self) -> str: |
| 169 | return self.http_response.http_version |
| 170 | |
| 171 | @property |
| 172 | def is_closed(self) -> bool: |
| 173 | return self.http_response.is_closed |
| 174 | |
| 175 | @property |
| 176 | def elapsed(self) -> datetime.timedelta: |
| 177 | """The time taken for the complete request/response cycle to complete.""" |
| 178 | return self.http_response.elapsed |
| 179 | |
| 180 | def _parse(self, *, to: type[_T] | None = None) -> R | _T: |
| 181 | # unwrap `Annotated[T, ...]` -> `T` |
| 182 | if to and is_annotated_type(to): |
| 183 | to = extract_type_arg(to, 0) |
| 184 | |
| 185 | if self._stream: |
| 186 | if to: |
| 187 | if not is_stream_class_type(to): |
| 188 | raise TypeError(f"Expected custom parse type to be a subclass of {Stream} or {AsyncStream}") |
| 189 | |
| 190 | return cast( |
| 191 | _T, |
| 192 | to( |
| 193 | cast_to=extract_stream_chunk_type( |
| 194 | to, |
| 195 | failure_message="Expected custom stream type to be passed with a type argument, e.g. Stream[ChunkType]", |
| 196 | ), |
| 197 | response=self.http_response, |
| 198 | client=cast(Any, self._client), |
| 199 | ), |
| 200 | ) |
| 201 | |
| 202 | if self._stream_cls: |
| 203 | return cast( |
| 204 | R, |
| 205 | self._stream_cls( |
| 206 | cast_to=extract_stream_chunk_type(self._stream_cls), |
| 207 | response=self.http_response, |
| 208 | client=cast(Any, self._client), |
| 209 | ), |
| 210 | ) |
| 211 | |
| 212 | stream_cls = cast("type[Stream[Any]] | type[AsyncStream[Any]] | None", self._client._default_stream_cls) |
| 213 | if stream_cls is None: |
| 214 | raise MissingStreamClassError() |
| 215 | |
| 216 | return cast( |
| 217 | R, |
| 218 | stream_cls( |
| 219 | cast_to=self._cast_to, |
| 220 | response=self.http_response, |
| 221 | client=cast(Any, self._client), |
| 222 | ), |
| 223 | ) |
| 224 | |
| 225 | cast_to = to if to is not None else self._cast_to |
| 226 | |
| 227 | # unwrap `Annotated[T, ...]` -> `T` |
| 228 | if is_annotated_type(cast_to): |
| 229 | cast_to = extract_type_arg(cast_to, 0) |
| 230 | |
| 231 | if cast_to is NoneType: |
| 232 | return cast(R, None) |
| 233 | |
| 234 | response = self.http_response |
| 235 | if cast_to == str: |
| 236 | return cast(R, response.text) |
| 237 | |
| 238 | if cast_to == int: |
| 239 | return cast(R, int(response.text)) |
| 240 | |
| 241 | if cast_to == float: |
| 242 | return cast(R, float(response.text)) |
| 243 | |
| 244 | origin = get_origin(cast_to) or cast_to |
| 245 | |
| 246 | if inspect.isclass(origin) and issubclass(origin, HttpxBinaryResponseContent): |
| 247 | return cast(R, cast_to(response)) # type: ignore |
| 248 | |
| 249 | if origin == LegacyAPIResponse: |
| 250 | raise RuntimeError("Unexpected state - cast_to is `APIResponse`") |
| 251 | |
| 252 | if inspect.isclass(origin) and issubclass(origin, httpx.Response): |
| 253 | # Because of the invariance of our ResponseT TypeVar, users can subclass httpx.Response |
| 254 | # and pass that class to our request functions. We cannot change the variance to be either |
| 255 | # covariant or contravariant as that makes our usage of ResponseT illegal. We could construct |
| 256 | # the response class ourselves but that is something that should be supported directly in httpx |
| 257 | # as it would be easy to incorrectly construct the Response object due to the multitude of arguments. |
| 258 | if cast_to != httpx.Response: |
| 259 | raise ValueError(f"Subclasses of httpx.Response cannot be passed to `cast_to`") |
| 260 | return cast(R, response) |
| 261 | |
| 262 | if inspect.isclass(origin) and not issubclass(origin, BaseModel) and issubclass(origin, pydantic.BaseModel): |
| 263 | raise TypeError("Pydantic models must subclass our base model type, e.g. `from openai import BaseModel`") |
| 264 | |
| 265 | if ( |
| 266 | cast_to is not object |
| 267 | and not origin is list |
| 268 | and not origin is dict |
| 269 | and not origin is Union |
| 270 | and not issubclass(origin, BaseModel) |
| 271 | ): |
| 272 | raise RuntimeError( |
| 273 | f"Unsupported type, expected {cast_to} to be a subclass of {BaseModel}, {dict}, {list}, {Union}, {NoneType}, {str} or {httpx.Response}." |
| 274 | ) |
| 275 | |
| 276 | # split is required to handle cases where additional information is included |
| 277 | # in the response, e.g. application/json; charset=utf-8 |
| 278 | content_type, *_ = response.headers.get("content-type", "*").split(";") |
| 279 | if content_type != "application/json": |
| 280 | if is_basemodel(cast_to): |
| 281 | try: |
| 282 | data = response.json() |
| 283 | except Exception as exc: |
| 284 | log.debug("Could not read JSON from response data due to %s - %s", type(exc), exc) |
| 285 | else: |
| 286 | return self._client._process_response_data( |
| 287 | data=data, |
| 288 | cast_to=cast_to, # type: ignore |
| 289 | response=response, |
| 290 | ) |
| 291 | |
| 292 | if self._client._strict_response_validation: |
| 293 | raise APIResponseValidationError( |
| 294 | response=response, |
| 295 | message=f"Expected Content-Type response header to be `application/json` but received `{content_type}` instead.", |
| 296 | body=response.text, |
| 297 | ) |
| 298 | |
| 299 | # If the API responds with content that isn't JSON then we just return |
| 300 | # the (decoded) text without performing any parsing so that you can still |
| 301 | # handle the response however you need to. |
| 302 | return response.text # type: ignore |
| 303 | |
| 304 | data = response.json() |
| 305 | |
| 306 | return self._client._process_response_data( |
| 307 | data=data, |
| 308 | cast_to=cast_to, # type: ignore |
| 309 | response=response, |
| 310 | ) |
| 311 | |
| 312 | @override |
| 313 | def __repr__(self) -> str: |
| 314 | return f"<APIResponse [{self.status_code} {self.http_response.reason_phrase}] type={self._cast_to}>" |
| 315 | |
| 316 | |
| 317 | class MissingStreamClassError(TypeError): |
| 318 | def __init__(self) -> None: |
| 319 | super().__init__( |
| 320 | "The `stream` argument was set to `True` but the `stream_cls` argument was not given. See `openai._streaming` for reference", |
| 321 | ) |
| 322 | |
| 323 | |
| 324 | def to_raw_response_wrapper(func: Callable[P, R]) -> Callable[P, LegacyAPIResponse[R]]: |
| 325 | """Higher order function that takes one of our bound API methods and wraps it |
| 326 | to support returning the raw `APIResponse` object directly. |
| 327 | """ |
| 328 | |
| 329 | @functools.wraps(func) |
| 330 | def wrapped(*args: P.args, **kwargs: P.kwargs) -> LegacyAPIResponse[R]: |
| 331 | extra_headers: dict[str, str] = {**(cast(Any, kwargs.get("extra_headers")) or {})} |
| 332 | extra_headers[RAW_RESPONSE_HEADER] = "true" |
| 333 | |
| 334 | kwargs["extra_headers"] = extra_headers |
| 335 | |
| 336 | return cast(LegacyAPIResponse[R], func(*args, **kwargs)) |
| 337 | |
| 338 | return wrapped |
| 339 | |
| 340 | |
| 341 | def async_to_raw_response_wrapper(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[LegacyAPIResponse[R]]]: |
| 342 | """Higher order function that takes one of our bound API methods and wraps it |
| 343 | to support returning the raw `APIResponse` object directly. |
| 344 | """ |
| 345 | |
| 346 | @functools.wraps(func) |
| 347 | async def wrapped(*args: P.args, **kwargs: P.kwargs) -> LegacyAPIResponse[R]: |
| 348 | extra_headers: dict[str, str] = {**(cast(Any, kwargs.get("extra_headers")) or {})} |
| 349 | extra_headers[RAW_RESPONSE_HEADER] = "true" |
| 350 | |
| 351 | kwargs["extra_headers"] = extra_headers |
| 352 | |
| 353 | return cast(LegacyAPIResponse[R], await func(*args, **kwargs)) |
| 354 | |
| 355 | return wrapped |
| 356 | |
| 357 | |
| 358 | class HttpxBinaryResponseContent: |
| 359 | response: httpx.Response |
| 360 | |
| 361 | def __init__(self, response: httpx.Response) -> None: |
| 362 | self.response = response |
| 363 | |
| 364 | @property |
| 365 | def content(self) -> bytes: |
| 366 | return self.response.content |
| 367 | |
| 368 | @property |
| 369 | def text(self) -> str: |
| 370 | return self.response.text |
| 371 | |
| 372 | @property |
| 373 | def encoding(self) -> str | None: |
| 374 | return self.response.encoding |
| 375 | |
| 376 | @property |
| 377 | def charset_encoding(self) -> str | None: |
| 378 | return self.response.charset_encoding |
| 379 | |
| 380 | def json(self, **kwargs: Any) -> Any: |
| 381 | return self.response.json(**kwargs) |
| 382 | |
| 383 | def read(self) -> bytes: |
| 384 | return self.response.read() |
| 385 | |
| 386 | def iter_bytes(self, chunk_size: int | None = None) -> Iterator[bytes]: |
| 387 | return self.response.iter_bytes(chunk_size) |
| 388 | |
| 389 | def iter_text(self, chunk_size: int | None = None) -> Iterator[str]: |
| 390 | return self.response.iter_text(chunk_size) |
| 391 | |
| 392 | def iter_lines(self) -> Iterator[str]: |
| 393 | return self.response.iter_lines() |
| 394 | |
| 395 | def iter_raw(self, chunk_size: int | None = None) -> Iterator[bytes]: |
| 396 | return self.response.iter_raw(chunk_size) |
| 397 | |
| 398 | def write_to_file( |
| 399 | self, |
| 400 | file: str | os.PathLike[str], |
| 401 | ) -> None: |
| 402 | """Write the output to the given file. |
| 403 | |
| 404 | Accepts a filename or any path-like object, e.g. pathlib.Path |
| 405 | |
| 406 | Note: if you want to stream the data to the file instead of writing |
| 407 | all at once then you should use `.with_streaming_response` when making |
| 408 | the API request, e.g. `client.with_streaming_response.foo().stream_to_file('my_filename.txt')` |
| 409 | """ |
| 410 | with open(file, mode="wb") as f: |
| 411 | for data in self.response.iter_bytes(): |
| 412 | f.write(data) |
| 413 | |
| 414 | @deprecated( |
| 415 | "Due to a bug, this method doesn't actually stream the response content, `.with_streaming_response.method()` should be used instead" |
| 416 | ) |
| 417 | def stream_to_file( |
| 418 | self, |
| 419 | file: str | os.PathLike[str], |
| 420 | *, |
| 421 | chunk_size: int | None = None, |
| 422 | ) -> None: |
| 423 | with open(file, mode="wb") as f: |
| 424 | for data in self.response.iter_bytes(chunk_size): |
| 425 | f.write(data) |
| 426 | |
| 427 | def close(self) -> None: |
| 428 | return self.response.close() |
| 429 | |
| 430 | async def aread(self) -> bytes: |
| 431 | return await self.response.aread() |
| 432 | |
| 433 | async def aiter_bytes(self, chunk_size: int | None = None) -> AsyncIterator[bytes]: |
| 434 | return self.response.aiter_bytes(chunk_size) |
| 435 | |
| 436 | async def aiter_text(self, chunk_size: int | None = None) -> AsyncIterator[str]: |
| 437 | return self.response.aiter_text(chunk_size) |
| 438 | |
| 439 | async def aiter_lines(self) -> AsyncIterator[str]: |
| 440 | return self.response.aiter_lines() |
| 441 | |
| 442 | async def aiter_raw(self, chunk_size: int | None = None) -> AsyncIterator[bytes]: |
| 443 | return self.response.aiter_raw(chunk_size) |
| 444 | |
| 445 | @deprecated( |
| 446 | "Due to a bug, this method doesn't actually stream the response content, `.with_streaming_response.method()` should be used instead" |
| 447 | ) |
| 448 | async def astream_to_file( |
| 449 | self, |
| 450 | file: str | os.PathLike[str], |
| 451 | *, |
| 452 | chunk_size: int | None = None, |
| 453 | ) -> None: |
| 454 | path = anyio.Path(file) |
| 455 | async with await path.open(mode="wb") as f: |
| 456 | async for data in self.response.aiter_bytes(chunk_size): |
| 457 | await f.write(data) |
| 458 | |
| 459 | async def aclose(self) -> None: |
| 460 | return await self.response.aclose() |
| 461 | |