1from __future__ import annotations
  2
  3import io
  4import os
  5import pathlib
  6from typing import overload
  7from typing_extensions import TypeGuard
  8
  9import anyio
 10
 11from ._types import (
 12    FileTypes,
 13    FileContent,
 14    RequestFiles,
 15    HttpxFileTypes,
 16    Base64FileInput,
 17    HttpxFileContent,
 18    HttpxRequestFiles,
 19)
 20from ._utils import is_tuple_t, is_mapping_t, is_sequence_t
 21
 22
 23def is_base64_file_input(obj: object) -> TypeGuard[Base64FileInput]:
 24    return isinstance(obj, io.IOBase) or isinstance(obj, os.PathLike)
 25
 26
 27def is_file_content(obj: object) -> TypeGuard[FileContent]:
 28    return (
 29        isinstance(obj, bytes) or isinstance(obj, tuple) or isinstance(obj, io.IOBase) or isinstance(obj, os.PathLike)
 30    )
 31
 32
 33def assert_is_file_content(obj: object, *, key: str | None = None) -> None:
 34    if not is_file_content(obj):
 35        prefix = f"Expected entry at `{key}`" if key is not None else f"Expected file input `{obj!r}`"
 36        raise RuntimeError(
 37            f"{prefix} to be bytes, an io.IOBase instance, PathLike or a tuple but received {type(obj)} instead. See https://github.com/openai/openai-python/tree/main#file-uploads"
 38        ) from None
 39
 40
 41@overload
 42def to_httpx_files(files: None) -> None: ...
 43
 44
 45@overload
 46def to_httpx_files(files: RequestFiles) -> HttpxRequestFiles: ...
 47
 48
 49def to_httpx_files(files: RequestFiles | None) -> HttpxRequestFiles | None:
 50    if files is None:
 51        return None
 52
 53    if is_mapping_t(files):
 54        files = {key: _transform_file(file) for key, file in files.items()}
 55    elif is_sequence_t(files):
 56        files = [(key, _transform_file(file)) for key, file in files]
 57    else:
 58        raise TypeError(f"Unexpected file type input {type(files)}, expected mapping or sequence")
 59
 60    return files
 61
 62
 63def _transform_file(file: FileTypes) -> HttpxFileTypes:
 64    if is_file_content(file):
 65        if isinstance(file, os.PathLike):
 66            path = pathlib.Path(file)
 67            return (path.name, path.read_bytes())
 68
 69        return file
 70
 71    if is_tuple_t(file):
 72        return (file[0], read_file_content(file[1]), *file[2:])
 73
 74    raise TypeError(f"Expected file types input to be a FileContent type or to be a tuple")
 75
 76
 77def read_file_content(file: FileContent) -> HttpxFileContent:
 78    if isinstance(file, os.PathLike):
 79        return pathlib.Path(file).read_bytes()
 80    return file
 81
 82
 83@overload
 84async def async_to_httpx_files(files: None) -> None: ...
 85
 86
 87@overload
 88async def async_to_httpx_files(files: RequestFiles) -> HttpxRequestFiles: ...
 89
 90
 91async def async_to_httpx_files(files: RequestFiles | None) -> HttpxRequestFiles | None:
 92    if files is None:
 93        return None
 94
 95    if is_mapping_t(files):
 96        files = {key: await _async_transform_file(file) for key, file in files.items()}
 97    elif is_sequence_t(files):
 98        files = [(key, await _async_transform_file(file)) for key, file in files]
 99    else:
100        raise TypeError("Unexpected file type input {type(files)}, expected mapping or sequence")
101
102    return files
103
104
105async def _async_transform_file(file: FileTypes) -> HttpxFileTypes:
106    if is_file_content(file):
107        if isinstance(file, os.PathLike):
108            path = anyio.Path(file)
109            return (path.name, await path.read_bytes())
110
111        return file
112
113    if is_tuple_t(file):
114        return (file[0], await async_read_file_content(file[1]), *file[2:])
115
116    raise TypeError(f"Expected file types input to be a FileContent type or to be a tuple")
117
118
119async def async_read_file_content(file: FileContent) -> HttpxFileContent:
120    if isinstance(file, os.PathLike):
121        return await anyio.Path(file).read_bytes()
122
123    return file