# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. from __future__ import annotations import time import typing_extensions from typing import Mapping, cast from typing_extensions import Literal import httpx from .. import _legacy_response from ..types import file_list_params, file_create_params from .._types import NOT_GIVEN, Body, Query, Headers, NotGiven, FileTypes from .._utils import ( extract_files, maybe_transform, deepcopy_minimal, async_maybe_transform, ) from .._compat import cached_property from .._resource import SyncAPIResource, AsyncAPIResource from .._response import ( StreamedBinaryAPIResponse, AsyncStreamedBinaryAPIResponse, to_streamed_response_wrapper, async_to_streamed_response_wrapper, to_custom_streamed_response_wrapper, async_to_custom_streamed_response_wrapper, ) from ..pagination import SyncPage, AsyncPage from .._base_client import ( AsyncPaginator, make_request_options, ) from ..types.file_object import FileObject from ..types.file_deleted import FileDeleted __all__ = ["Files", "AsyncFiles"] class Files(SyncAPIResource): @cached_property def with_raw_response(self) -> FilesWithRawResponse: return FilesWithRawResponse(self) @cached_property def with_streaming_response(self) -> FilesWithStreamingResponse: return FilesWithStreamingResponse(self) def create( self, *, file: FileTypes, purpose: Literal["fine-tune", "assistants"], # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileObject: """Upload a file that can be used across various endpoints. The size of all the files uploaded by one organization can be up to 100 GB. The size of individual files can be a maximum of 512 MB or 2 million tokens for Assistants. See the [Assistants Tools guide](https://platform.openai.com/docs/assistants/tools) to learn more about the types of files supported. The Fine-tuning API only supports `.jsonl` files. Please [contact us](https://help.openai.com/) if you need to increase these storage limits. Args: file: The File object (not file name) to be uploaded. purpose: The intended purpose of the uploaded file. Use "fine-tune" for [Fine-tuning](https://platform.openai.com/docs/api-reference/fine-tuning) and "assistants" for [Assistants](https://platform.openai.com/docs/api-reference/assistants) and [Messages](https://platform.openai.com/docs/api-reference/messages). This allows us to validate the format of the uploaded file is correct for fine-tuning. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ body = deepcopy_minimal( { "file": file, "purpose": purpose, } ) files = extract_files(cast(Mapping[str, object], body), paths=[["file"]]) if files: # It should be noted that the actual Content-Type header that will be # sent to the server will contain a `boundary` parameter, e.g. # multipart/form-data; boundary=---abc-- extra_headers = {"Content-Type": "multipart/form-data", **(extra_headers or {})} return self._post( "/files", body=maybe_transform(body, file_create_params.FileCreateParams), files=files, options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileObject, ) def retrieve( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileObject: """ Returns information about a specific file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return self._get( f"/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileObject, ) def list( self, *, purpose: str | NotGiven = NOT_GIVEN, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> SyncPage[FileObject]: """ Returns a list of files that belong to the user's organization. Args: purpose: Only return files with the given purpose. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ return self._get_api_list( "/files", page=SyncPage[FileObject], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform({"purpose": purpose}, file_list_params.FileListParams), ), model=FileObject, ) def delete( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileDeleted: """ Delete a file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return self._delete( f"/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileDeleted, ) def content( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> _legacy_response.HttpxBinaryResponseContent: """ Returns the contents of the specified file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"Accept": "application/binary", **(extra_headers or {})} return self._get( f"/files/{file_id}/content", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=_legacy_response.HttpxBinaryResponseContent, ) @typing_extensions.deprecated("The `.content()` method should be used instead") def retrieve_content( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> str: """ Returns the contents of the specified file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return self._get( f"/files/{file_id}/content", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=str, ) def wait_for_processing( self, id: str, *, poll_interval: float = 5.0, max_wait_seconds: float = 30 * 60, ) -> FileObject: """Waits for the given file to be processed, default timeout is 30 mins.""" TERMINAL_STATES = {"processed", "error", "deleted"} start = time.time() file = self.retrieve(id) while file.status not in TERMINAL_STATES: self._sleep(poll_interval) file = self.retrieve(id) if time.time() - start > max_wait_seconds: raise RuntimeError( f"Giving up on waiting for file {id} to finish processing after {max_wait_seconds} seconds." ) return file class AsyncFiles(AsyncAPIResource): @cached_property def with_raw_response(self) -> AsyncFilesWithRawResponse: return AsyncFilesWithRawResponse(self) @cached_property def with_streaming_response(self) -> AsyncFilesWithStreamingResponse: return AsyncFilesWithStreamingResponse(self) async def create( self, *, file: FileTypes, purpose: Literal["fine-tune", "assistants"], # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileObject: """Upload a file that can be used across various endpoints. The size of all the files uploaded by one organization can be up to 100 GB. The size of individual files can be a maximum of 512 MB or 2 million tokens for Assistants. See the [Assistants Tools guide](https://platform.openai.com/docs/assistants/tools) to learn more about the types of files supported. The Fine-tuning API only supports `.jsonl` files. Please [contact us](https://help.openai.com/) if you need to increase these storage limits. Args: file: The File object (not file name) to be uploaded. purpose: The intended purpose of the uploaded file. Use "fine-tune" for [Fine-tuning](https://platform.openai.com/docs/api-reference/fine-tuning) and "assistants" for [Assistants](https://platform.openai.com/docs/api-reference/assistants) and [Messages](https://platform.openai.com/docs/api-reference/messages). This allows us to validate the format of the uploaded file is correct for fine-tuning. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ body = deepcopy_minimal( { "file": file, "purpose": purpose, } ) files = extract_files(cast(Mapping[str, object], body), paths=[["file"]]) if files: # It should be noted that the actual Content-Type header that will be # sent to the server will contain a `boundary` parameter, e.g. # multipart/form-data; boundary=---abc-- extra_headers = {"Content-Type": "multipart/form-data", **(extra_headers or {})} return await self._post( "/files", body=await async_maybe_transform(body, file_create_params.FileCreateParams), files=files, options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileObject, ) async def retrieve( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileObject: """ Returns information about a specific file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return await self._get( f"/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileObject, ) def list( self, *, purpose: str | NotGiven = NOT_GIVEN, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> AsyncPaginator[FileObject, AsyncPage[FileObject]]: """ Returns a list of files that belong to the user's organization. Args: purpose: Only return files with the given purpose. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ return self._get_api_list( "/files", page=AsyncPage[FileObject], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform({"purpose": purpose}, file_list_params.FileListParams), ), model=FileObject, ) async def delete( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> FileDeleted: """ Delete a file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return await self._delete( f"/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileDeleted, ) async def content( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> _legacy_response.HttpxBinaryResponseContent: """ Returns the contents of the specified file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"Accept": "application/binary", **(extra_headers or {})} return await self._get( f"/files/{file_id}/content", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=_legacy_response.HttpxBinaryResponseContent, ) @typing_extensions.deprecated("The `.content()` method should be used instead") async def retrieve_content( self, file_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN, ) -> str: """ Returns the contents of the specified file. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") return await self._get( f"/files/{file_id}/content", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=str, ) async def wait_for_processing( self, id: str, *, poll_interval: float = 5.0, max_wait_seconds: float = 30 * 60, ) -> FileObject: """Waits for the given file to be processed, default timeout is 30 mins.""" TERMINAL_STATES = {"processed", "error", "deleted"} start = time.time() file = await self.retrieve(id) while file.status not in TERMINAL_STATES: await self._sleep(poll_interval) file = await self.retrieve(id) if time.time() - start > max_wait_seconds: raise RuntimeError( f"Giving up on waiting for file {id} to finish processing after {max_wait_seconds} seconds." ) return file class FilesWithRawResponse: def __init__(self, files: Files) -> None: self._files = files self.create = _legacy_response.to_raw_response_wrapper( files.create, ) self.retrieve = _legacy_response.to_raw_response_wrapper( files.retrieve, ) self.list = _legacy_response.to_raw_response_wrapper( files.list, ) self.delete = _legacy_response.to_raw_response_wrapper( files.delete, ) self.content = _legacy_response.to_raw_response_wrapper( files.content, ) self.retrieve_content = ( # pyright: ignore[reportDeprecated] _legacy_response.to_raw_response_wrapper( files.retrieve_content # pyright: ignore[reportDeprecated], ) ) class AsyncFilesWithRawResponse: def __init__(self, files: AsyncFiles) -> None: self._files = files self.create = _legacy_response.async_to_raw_response_wrapper( files.create, ) self.retrieve = _legacy_response.async_to_raw_response_wrapper( files.retrieve, ) self.list = _legacy_response.async_to_raw_response_wrapper( files.list, ) self.delete = _legacy_response.async_to_raw_response_wrapper( files.delete, ) self.content = _legacy_response.async_to_raw_response_wrapper( files.content, ) self.retrieve_content = ( # pyright: ignore[reportDeprecated] _legacy_response.async_to_raw_response_wrapper( files.retrieve_content # pyright: ignore[reportDeprecated], ) ) class FilesWithStreamingResponse: def __init__(self, files: Files) -> None: self._files = files self.create = to_streamed_response_wrapper( files.create, ) self.retrieve = to_streamed_response_wrapper( files.retrieve, ) self.list = to_streamed_response_wrapper( files.list, ) self.delete = to_streamed_response_wrapper( files.delete, ) self.content = to_custom_streamed_response_wrapper( files.content, StreamedBinaryAPIResponse, ) self.retrieve_content = ( # pyright: ignore[reportDeprecated] to_streamed_response_wrapper( files.retrieve_content # pyright: ignore[reportDeprecated], ) ) class AsyncFilesWithStreamingResponse: def __init__(self, files: AsyncFiles) -> None: self._files = files self.create = async_to_streamed_response_wrapper( files.create, ) self.retrieve = async_to_streamed_response_wrapper( files.retrieve, ) self.list = async_to_streamed_response_wrapper( files.list, ) self.delete = async_to_streamed_response_wrapper( files.delete, ) self.content = async_to_custom_streamed_response_wrapper( files.content, AsyncStreamedBinaryAPIResponse, ) self.retrieve_content = ( # pyright: ignore[reportDeprecated] async_to_streamed_response_wrapper( files.retrieve_content # pyright: ignore[reportDeprecated], ) )