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
0005b7c9d2f9965540d59071982cdf6fa42da26c

Branches

Tags

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

Clone

HTTPS

Download ZIP

openai-python/src/openai/_utils

src/openai/_utils/_utils.py

422lines · modecode

RawDownload
Latest commit unavailable.
1from __future__ import annotations
2
3import os
4import re
5import inspect
6import functools
7from typing import (
8 Any,
9 Tuple,
10 Mapping,
11 TypeVar,
12 Callable,
13 Iterable,
14 Sequence,
15 cast,
16 overload,
17)
18from pathlib import Path
19from datetime import date, datetime
20from typing_extensions import TypeGuard
21
22import sniffio
23
24from .._types import NotGiven, FileTypes, NotGivenOr, HeadersLike
25from .._compat import parse_date as parse_date, parse_datetime as parse_datetime
26
27_T = TypeVar("_T")
28_TupleT = TypeVar("_TupleT", bound=Tuple[object, ...])
29_MappingT = TypeVar("_MappingT", bound=Mapping[str, object])
30_SequenceT = TypeVar("_SequenceT", bound=Sequence[object])
31CallableT = TypeVar("CallableT", bound=Callable[..., Any])
32
33
34def flatten(t: Iterable[Iterable[_T]]) -> list[_T]:
35 return [item for sublist in t for item in sublist]
36
37
38def extract_files(
39 # TODO: this needs to take Dict but variance issues.....
40 # create protocol type ?
41 query: Mapping[str, object],
42 *,
43 paths: Sequence[Sequence[str]],
44) -> list[tuple[str, FileTypes]]:
45 """Recursively extract files from the given dictionary based on specified paths.
46
47 A path may look like this ['foo', 'files', '<array>', 'data'].
48
49 Note: this mutates the given dictionary.
50 """
51 files: list[tuple[str, FileTypes]] = []
52 for path in paths:
53 files.extend(_extract_items(query, path, index=0, flattened_key=None))
54 return files
55
56
57def _extract_items(
58 obj: object,
59 path: Sequence[str],
60 *,
61 index: int,
62 flattened_key: str | None,
63) -> list[tuple[str, FileTypes]]:
64 try:
65 key = path[index]
66 except IndexError:
67 if isinstance(obj, NotGiven):
68 # no value was provided - we can safely ignore
69 return []
70
71 # cyclical import
72 from .._files import assert_is_file_content
73
74 # We have exhausted the path, return the entry we found.
75 assert flattened_key is not None
76
77 if is_list(obj):
78 files: list[tuple[str, FileTypes]] = []
79 for entry in obj:
80 assert_is_file_content(entry, key=flattened_key + "[]" if flattened_key else "")
81 files.append((flattened_key + "[]", cast(FileTypes, entry)))
82 return files
83
84 assert_is_file_content(obj, key=flattened_key)
85 return [(flattened_key, cast(FileTypes, obj))]
86
87 index += 1
88 if is_dict(obj):
89 try:
90 # We are at the last entry in the path so we must remove the field
91 if (len(path)) == index:
92 item = obj.pop(key)
93 else:
94 item = obj[key]
95 except KeyError:
96 # Key was not present in the dictionary, this is not indicative of an error
97 # as the given path may not point to a required field. We also do not want
98 # to enforce required fields as the API may differ from the spec in some cases.
99 return []
100 if flattened_key is None:
101 flattened_key = key
102 else:
103 flattened_key += f"[{key}]"
104 return _extract_items(
105 item,
106 path,
107 index=index,
108 flattened_key=flattened_key,
109 )
110 elif is_list(obj):
111 if key != "<array>":
112 return []
113
114 return flatten(
115 [
116 _extract_items(
117 item,
118 path,
119 index=index,
120 flattened_key=flattened_key + "[]" if flattened_key is not None else "[]",
121 )
122 for item in obj
123 ]
124 )
125
126 # Something unexpected was passed, just ignore it.
127 return []
128
129
130def is_given(obj: NotGivenOr[_T]) -> TypeGuard[_T]:
131 return not isinstance(obj, NotGiven)
132
133
134# Type safe methods for narrowing types with TypeVars.
135# The default narrowing for isinstance(obj, dict) is dict[unknown, unknown],
136# however this cause Pyright to rightfully report errors. As we know we don't
137# care about the contained types we can safely use `object` in it's place.
138#
139# There are two separate functions defined, `is_*` and `is_*_t` for different use cases.
140# `is_*` is for when you're dealing with an unknown input
141# `is_*_t` is for when you're narrowing a known union type to a specific subset
142
143
144def is_tuple(obj: object) -> TypeGuard[tuple[object, ...]]:
145 return isinstance(obj, tuple)
146
147
148def is_tuple_t(obj: _TupleT | object) -> TypeGuard[_TupleT]:
149 return isinstance(obj, tuple)
150
151
152def is_sequence(obj: object) -> TypeGuard[Sequence[object]]:
153 return isinstance(obj, Sequence)
154
155
156def is_sequence_t(obj: _SequenceT | object) -> TypeGuard[_SequenceT]:
157 return isinstance(obj, Sequence)
158
159
160def is_mapping(obj: object) -> TypeGuard[Mapping[str, object]]:
161 return isinstance(obj, Mapping)
162
163
164def is_mapping_t(obj: _MappingT | object) -> TypeGuard[_MappingT]:
165 return isinstance(obj, Mapping)
166
167
168def is_dict(obj: object) -> TypeGuard[dict[object, object]]:
169 return isinstance(obj, dict)
170
171
172def is_list(obj: object) -> TypeGuard[list[object]]:
173 return isinstance(obj, list)
174
175
176def is_iterable(obj: object) -> TypeGuard[Iterable[object]]:
177 return isinstance(obj, Iterable)
178
179
180def deepcopy_minimal(item: _T) -> _T:
181 """Minimal reimplementation of copy.deepcopy() that will only copy certain object types:
182
183 - mappings, e.g. `dict`
184 - list
185
186 This is done for performance reasons.
187 """
188 if is_mapping(item):
189 return cast(_T, {k: deepcopy_minimal(v) for k, v in item.items()})
190 if is_list(item):
191 return cast(_T, [deepcopy_minimal(entry) for entry in item])
192 return item
193
194
195# copied from https://github.com/Rapptz/RoboDanny
196def human_join(seq: Sequence[str], *, delim: str = ", ", final: str = "or") -> str:
197 size = len(seq)
198 if size == 0:
199 return ""
200
201 if size == 1:
202 return seq[0]
203
204 if size == 2:
205 return f"{seq[0]} {final} {seq[1]}"
206
207 return delim.join(seq[:-1]) + f" {final} {seq[-1]}"
208
209
210def quote(string: str) -> str:
211 """Add single quotation marks around the given string. Does *not* do any escaping."""
212 return f"'{string}'"
213
214
215def required_args(*variants: Sequence[str]) -> Callable[[CallableT], CallableT]:
216 """Decorator to enforce a given set of arguments or variants of arguments are passed to the decorated function.
217
218 Useful for enforcing runtime validation of overloaded functions.
219
220 Example usage:
221 ```py
222 @overload
223 def foo(*, a: str) -> str: ...
224
225
226 @overload
227 def foo(*, b: bool) -> str: ...
228
229
230 # This enforces the same constraints that a static type checker would
231 # i.e. that either a or b must be passed to the function
232 @required_args(["a"], ["b"])
233 def foo(*, a: str | None = None, b: bool | None = None) -> str: ...
234 ```
235 """
236
237 def inner(func: CallableT) -> CallableT:
238 params = inspect.signature(func).parameters
239 positional = [
240 name
241 for name, param in params.items()
242 if param.kind
243 in {
244 param.POSITIONAL_ONLY,
245 param.POSITIONAL_OR_KEYWORD,
246 }
247 ]
248
249 @functools.wraps(func)
250 def wrapper(*args: object, **kwargs: object) -> object:
251 given_params: set[str] = set()
252 for i, _ in enumerate(args):
253 try:
254 given_params.add(positional[i])
255 except IndexError:
256 raise TypeError(
257 f"{func.__name__}() takes {len(positional)} argument(s) but {len(args)} were given"
258 ) from None
259
260 for key in kwargs.keys():
261 given_params.add(key)
262
263 for variant in variants:
264 matches = all((param in given_params for param in variant))
265 if matches:
266 break
267 else: # no break
268 if len(variants) > 1:
269 variations = human_join(
270 ["(" + human_join([quote(arg) for arg in variant], final="and") + ")" for variant in variants]
271 )
272 msg = f"Missing required arguments; Expected either {variations} arguments to be given"
273 else:
274 assert len(variants) > 0
275
276 # TODO: this error message is not deterministic
277 missing = list(set(variants[0]) - given_params)
278 if len(missing) > 1:
279 msg = f"Missing required arguments: {human_join([quote(arg) for arg in missing])}"
280 else:
281 msg = f"Missing required argument: {quote(missing[0])}"
282 raise TypeError(msg)
283 return func(*args, **kwargs)
284
285 return wrapper # type: ignore
286
287 return inner
288
289
290_K = TypeVar("_K")
291_V = TypeVar("_V")
292
293
294@overload
295def strip_not_given(obj: None) -> None: ...
296
297
298@overload
299def strip_not_given(obj: Mapping[_K, _V | NotGiven]) -> dict[_K, _V]: ...
300
301
302@overload
303def strip_not_given(obj: object) -> object: ...
304
305
306def strip_not_given(obj: object | None) -> object:
307 """Remove all top-level keys where their values are instances of `NotGiven`"""
308 if obj is None:
309 return None
310
311 if not is_mapping(obj):
312 return obj
313
314 return {key: value for key, value in obj.items() if not isinstance(value, NotGiven)}
315
316
317def coerce_integer(val: str) -> int:
318 return int(val, base=10)
319
320
321def coerce_float(val: str) -> float:
322 return float(val)
323
324
325def coerce_boolean(val: str) -> bool:
326 return val == "true" or val == "1" or val == "on"
327
328
329def maybe_coerce_integer(val: str | None) -> int | None:
330 if val is None:
331 return None
332 return coerce_integer(val)
333
334
335def maybe_coerce_float(val: str | None) -> float | None:
336 if val is None:
337 return None
338 return coerce_float(val)
339
340
341def maybe_coerce_boolean(val: str | None) -> bool | None:
342 if val is None:
343 return None
344 return coerce_boolean(val)
345
346
347def removeprefix(string: str, prefix: str) -> str:
348 """Remove a prefix from a string.
349
350 Backport of `str.removeprefix` for Python < 3.9
351 """
352 if string.startswith(prefix):
353 return string[len(prefix) :]
354 return string
355
356
357def removesuffix(string: str, suffix: str) -> str:
358 """Remove a suffix from a string.
359
360 Backport of `str.removesuffix` for Python < 3.9
361 """
362 if string.endswith(suffix):
363 return string[: -len(suffix)]
364 return string
365
366
367def file_from_path(path: str) -> FileTypes:
368 contents = Path(path).read_bytes()
369 file_name = os.path.basename(path)
370 return (file_name, contents)
371
372
373def get_required_header(headers: HeadersLike, header: str) -> str:
374 lower_header = header.lower()
375 if is_mapping_t(headers):
376 # mypy doesn't understand the type narrowing here
377 for k, v in headers.items(): # type: ignore
378 if k.lower() == lower_header and isinstance(v, str):
379 return v
380
381 # to deal with the case where the header looks like Stainless-Event-Id
382 intercaps_header = re.sub(r"([^\w])(\w)", lambda pat: pat.group(1) + pat.group(2).upper(), header.capitalize())
383
384 for normalized_header in [header, lower_header, header.upper(), intercaps_header]:
385 value = headers.get(normalized_header)
386 if value:
387 return value
388
389 raise ValueError(f"Could not find {header} header")
390
391
392def get_async_library() -> str:
393 try:
394 return sniffio.current_async_library()
395 except Exception:
396 return "false"
397
398
399def lru_cache(*, maxsize: int | None = 128) -> Callable[[CallableT], CallableT]:
400 """A version of functools.lru_cache that retains the type signature
401 for the wrapped function arguments.
402 """
403 wrapper = functools.lru_cache( # noqa: TID251
404 maxsize=maxsize,
405 )
406 return cast(Any, wrapper) # type: ignore[no-any-return]
407
408
409def json_safe(data: object) -> object:
410 """Translates a mapping / sequence recursively in the same fashion
411 as `pydantic` v2's `model_dump(mode="json")`.
412 """
413 if is_mapping(data):
414 return {json_safe(key): json_safe(value) for key, value in data.items()}
415
416 if is_iterable(data) and not isinstance(data, (str, bytes, bytearray)):
417 return [json_safe(item) for item in data]
418
419 if isinstance(data, (datetime, date)):
420 return data.isoformat()
421
422 return data
423