main
1from __future__ import annotations
2
3from os import PathLike
4from typing import (
5 IO,
6 TYPE_CHECKING,
7 Any,
8 Dict,
9 List,
10 Type,
11 Tuple,
12 Union,
13 Mapping,
14 TypeVar,
15 Callable,
16 Iterator,
17 Optional,
18 Sequence,
19)
20from typing_extensions import (
21 Set,
22 Literal,
23 Protocol,
24 TypeAlias,
25 TypedDict,
26 SupportsIndex,
27 overload,
28 override,
29 runtime_checkable,
30)
31
32import httpx
33import pydantic
34from httpx import URL, Proxy, Timeout, Response, BaseTransport, AsyncBaseTransport
35
36if TYPE_CHECKING:
37 from ._models import BaseModel
38 from ._response import APIResponse, AsyncAPIResponse
39 from ._legacy_response import HttpxBinaryResponseContent
40
41Transport = BaseTransport
42AsyncTransport = AsyncBaseTransport
43Query = Mapping[str, object]
44Body = object
45AnyMapping = Mapping[str, object]
46ModelT = TypeVar("ModelT", bound=pydantic.BaseModel)
47_T = TypeVar("_T")
48
49
50# Approximates httpx internal ProxiesTypes and RequestFiles types
51# while adding support for `PathLike` instances
52ProxiesDict = Dict["str | URL", Union[None, str, URL, Proxy]]
53ProxiesTypes = Union[str, Proxy, ProxiesDict]
54if TYPE_CHECKING:
55 Base64FileInput = Union[IO[bytes], PathLike[str]]
56 FileContent = Union[IO[bytes], bytes, PathLike[str]]
57else:
58 Base64FileInput = Union[IO[bytes], PathLike]
59 FileContent = Union[IO[bytes], bytes, PathLike] # PathLike is not subscriptable in Python 3.8.
60FileTypes = Union[
61 # file (or bytes)
62 FileContent,
63 # (filename, file (or bytes))
64 Tuple[Optional[str], FileContent],
65 # (filename, file (or bytes), content_type)
66 Tuple[Optional[str], FileContent, Optional[str]],
67 # (filename, file (or bytes), content_type, headers)
68 Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]],
69]
70RequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]
71
72# duplicate of the above but without our custom file support
73HttpxFileContent = Union[IO[bytes], bytes]
74HttpxFileTypes = Union[
75 # file (or bytes)
76 HttpxFileContent,
77 # (filename, file (or bytes))
78 Tuple[Optional[str], HttpxFileContent],
79 # (filename, file (or bytes), content_type)
80 Tuple[Optional[str], HttpxFileContent, Optional[str]],
81 # (filename, file (or bytes), content_type, headers)
82 Tuple[Optional[str], HttpxFileContent, Optional[str], Mapping[str, str]],
83]
84HttpxRequestFiles = Union[Mapping[str, HttpxFileTypes], Sequence[Tuple[str, HttpxFileTypes]]]
85
86# Workaround to support (cast_to: Type[ResponseT]) -> ResponseT
87# where ResponseT includes `None`. In order to support directly
88# passing `None`, overloads would have to be defined for every
89# method that uses `ResponseT` which would lead to an unacceptable
90# amount of code duplication and make it unreadable. See _base_client.py
91# for example usage.
92#
93# This unfortunately means that you will either have
94# to import this type and pass it explicitly:
95#
96# from openai import NoneType
97# client.get('/foo', cast_to=NoneType)
98#
99# or build it yourself:
100#
101# client.get('/foo', cast_to=type(None))
102if TYPE_CHECKING:
103 NoneType: Type[None]
104else:
105 NoneType = type(None)
106
107
108class RequestOptions(TypedDict, total=False):
109 headers: Headers
110 max_retries: int
111 timeout: float | Timeout | None
112 params: Query
113 extra_json: AnyMapping
114 idempotency_key: str
115 follow_redirects: bool
116
117
118# Sentinel class used until PEP 0661 is accepted
119class NotGiven:
120 """
121 For parameters with a meaningful None value, we need to distinguish between
122 the user explicitly passing None, and the user not passing the parameter at
123 all.
124
125 User code shouldn't need to use not_given directly.
126
127 For example:
128
129 ```py
130 def create(timeout: Timeout | None | NotGiven = not_given): ...
131
132
133 create(timeout=1) # 1s timeout
134 create(timeout=None) # No timeout
135 create() # Default timeout behavior
136 ```
137 """
138
139 def __bool__(self) -> Literal[False]:
140 return False
141
142 @override
143 def __repr__(self) -> str:
144 return "NOT_GIVEN"
145
146
147not_given = NotGiven()
148# for backwards compatibility:
149NOT_GIVEN = NotGiven()
150
151
152class Omit:
153 """
154 To explicitly omit something from being sent in a request, use `omit`.
155
156 ```py
157 # as the default `Content-Type` header is `application/json` that will be sent
158 client.post("/upload/files", files={"file": b"my raw file content"})
159
160 # you can't explicitly override the header as it has to be dynamically generated
161 # to look something like: 'multipart/form-data; boundary=0d8382fcf5f8c3be01ca2e11002d2983'
162 client.post(..., headers={"Content-Type": "multipart/form-data"})
163
164 # instead you can remove the default `application/json` header by passing omit
165 client.post(..., headers={"Content-Type": omit})
166 ```
167 """
168
169 def __bool__(self) -> Literal[False]:
170 return False
171
172
173omit = Omit()
174
175Omittable = Union[_T, Omit]
176
177
178@runtime_checkable
179class ModelBuilderProtocol(Protocol):
180 @classmethod
181 def build(
182 cls: type[_T],
183 *,
184 response: Response,
185 data: object,
186 ) -> _T: ...
187
188
189Headers = Mapping[str, Union[str, Omit]]
190
191
192class HeadersLikeProtocol(Protocol):
193 def get(self, __key: str) -> str | None: ...
194
195
196HeadersLike = Union[Headers, HeadersLikeProtocol]
197
198ResponseT = TypeVar(
199 "ResponseT",
200 bound=Union[
201 object,
202 str,
203 None,
204 "BaseModel",
205 List[Any],
206 Dict[str, Any],
207 Response,
208 ModelBuilderProtocol,
209 "APIResponse[Any]",
210 "AsyncAPIResponse[Any]",
211 "HttpxBinaryResponseContent",
212 ],
213)
214
215StrBytesIntFloat = Union[str, bytes, int, float]
216
217# Note: copied from Pydantic
218# https://github.com/pydantic/pydantic/blob/6f31f8f68ef011f84357330186f603ff295312fd/pydantic/main.py#L79
219IncEx: TypeAlias = Union[Set[int], Set[str], Mapping[int, Union["IncEx", bool]], Mapping[str, Union["IncEx", bool]]]
220
221PostParser = Callable[[Any], Any]
222
223
224@runtime_checkable
225class InheritsGeneric(Protocol):
226 """Represents a type that has inherited from `Generic`
227
228 The `__orig_bases__` property can be used to determine the resolved
229 type variable for a given base class.
230 """
231
232 __orig_bases__: tuple[_GenericAlias]
233
234
235class _GenericAlias(Protocol):
236 __origin__: type[object]
237
238
239class HttpxSendArgs(TypedDict, total=False):
240 auth: httpx.Auth
241 follow_redirects: bool
242
243
244_T_co = TypeVar("_T_co", covariant=True)
245
246
247if TYPE_CHECKING:
248 # This works because str.__contains__ does not accept object (either in typeshed or at runtime)
249 # https://github.com/hauntsaninja/useful_types/blob/5e9710f3875107d068e7679fd7fec9cfab0eff3b/useful_types/__init__.py#L285
250 class SequenceNotStr(Protocol[_T_co]):
251 @overload
252 def __getitem__(self, index: SupportsIndex, /) -> _T_co: ...
253 @overload
254 def __getitem__(self, index: slice, /) -> Sequence[_T_co]: ...
255 def __contains__(self, value: object, /) -> bool: ...
256 def __len__(self) -> int: ...
257 def __iter__(self) -> Iterator[_T_co]: ...
258 def index(self, value: Any, start: int = 0, stop: int = ..., /) -> int: ...
259 def count(self, value: Any, /) -> int: ...
260 def __reversed__(self) -> Iterator[_T_co]: ...
261else:
262 # just point this to a normal `Sequence` at runtime to avoid having to special case
263 # deserializing our custom sequence type
264 SequenceNotStr = Sequence