# File generated from our OpenAPI spec by Stainless. from __future__ import annotations from typing_extensions import Literal import httpx from .... import _legacy_response from ...._types import NOT_GIVEN, Body, Query, Headers, NotGiven from ...._utils import ( maybe_transform, async_maybe_transform, ) from ...._compat import cached_property from ...._resource import SyncAPIResource, AsyncAPIResource from ...._response import to_streamed_response_wrapper, async_to_streamed_response_wrapper from ....pagination import SyncCursorPage, AsyncCursorPage from ...._base_client import ( AsyncPaginator, make_request_options, ) from ....types.beta.assistants import AssistantFile, FileDeleteResponse, file_list_params, file_create_params __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, assistant_id: str, *, 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, ) -> AssistantFile: """ Create an assistant file by attaching a [File](https://platform.openai.com/docs/api-reference/files) to an [assistant](https://platform.openai.com/docs/api-reference/assistants). Args: file_id: A [File](https://platform.openai.com/docs/api-reference/files) ID (with `purpose="assistants"`) that the assistant should use. Useful for tools like `retrieval` and `code_interpreter` that can access files. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return self._post( f"/assistants/{assistant_id}/files", body=maybe_transform({"file_id": file_id}, file_create_params.FileCreateParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=AssistantFile, ) def retrieve( self, file_id: str, *, assistant_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, ) -> AssistantFile: """ Retrieves an AssistantFile. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return self._get( f"/assistants/{assistant_id}/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=AssistantFile, ) def list( self, assistant_id: str, *, after: str | NotGiven = NOT_GIVEN, before: str | NotGiven = NOT_GIVEN, limit: int | NotGiven = NOT_GIVEN, order: Literal["asc", "desc"] | 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, ) -> SyncCursorPage[AssistantFile]: """ Returns a list of assistant files. Args: after: A cursor for use in pagination. `after` is an object ID that defines your place in the list. For instance, if you make a list request and receive 100 objects, ending with obj_foo, your subsequent call can include after=obj_foo in order to fetch the next page of the list. before: A cursor for use in pagination. `before` is an object ID that defines your place in the list. For instance, if you make a list request and receive 100 objects, ending with obj_foo, your subsequent call can include before=obj_foo in order to fetch the previous page of the list. limit: A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 20. order: Sort order by the `created_at` timestamp of the objects. `asc` for ascending order and `desc` for descending order. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return self._get_api_list( f"/assistants/{assistant_id}/files", page=SyncCursorPage[AssistantFile], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform( { "after": after, "before": before, "limit": limit, "order": order, }, file_list_params.FileListParams, ), ), model=AssistantFile, ) def delete( self, file_id: str, *, assistant_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, ) -> FileDeleteResponse: """ Delete an assistant 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return self._delete( f"/assistants/{assistant_id}/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileDeleteResponse, ) 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, assistant_id: str, *, 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, ) -> AssistantFile: """ Create an assistant file by attaching a [File](https://platform.openai.com/docs/api-reference/files) to an [assistant](https://platform.openai.com/docs/api-reference/assistants). Args: file_id: A [File](https://platform.openai.com/docs/api-reference/files) ID (with `purpose="assistants"`) that the assistant should use. Useful for tools like `retrieval` and `code_interpreter` that can access files. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return await self._post( f"/assistants/{assistant_id}/files", body=await async_maybe_transform({"file_id": file_id}, file_create_params.FileCreateParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=AssistantFile, ) async def retrieve( self, file_id: str, *, assistant_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, ) -> AssistantFile: """ Retrieves an AssistantFile. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return await self._get( f"/assistants/{assistant_id}/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=AssistantFile, ) def list( self, assistant_id: str, *, after: str | NotGiven = NOT_GIVEN, before: str | NotGiven = NOT_GIVEN, limit: int | NotGiven = NOT_GIVEN, order: Literal["asc", "desc"] | 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[AssistantFile, AsyncCursorPage[AssistantFile]]: """ Returns a list of assistant files. Args: after: A cursor for use in pagination. `after` is an object ID that defines your place in the list. For instance, if you make a list request and receive 100 objects, ending with obj_foo, your subsequent call can include after=obj_foo in order to fetch the next page of the list. before: A cursor for use in pagination. `before` is an object ID that defines your place in the list. For instance, if you make a list request and receive 100 objects, ending with obj_foo, your subsequent call can include before=obj_foo in order to fetch the previous page of the list. limit: A limit on the number of objects to be returned. Limit can range between 1 and 100, and the default is 20. order: Sort order by the `created_at` timestamp of the objects. `asc` for ascending order and `desc` for descending order. 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return self._get_api_list( f"/assistants/{assistant_id}/files", page=AsyncCursorPage[AssistantFile], options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout, query=maybe_transform( { "after": after, "before": before, "limit": limit, "order": order, }, file_list_params.FileListParams, ), ), model=AssistantFile, ) async def delete( self, file_id: str, *, assistant_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, ) -> FileDeleteResponse: """ Delete an assistant 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 assistant_id: raise ValueError(f"Expected a non-empty value for `assistant_id` but received {assistant_id!r}") if not file_id: raise ValueError(f"Expected a non-empty value for `file_id` but received {file_id!r}") extra_headers = {"OpenAI-Beta": "assistants=v1", **(extra_headers or {})} return await self._delete( f"/assistants/{assistant_id}/files/{file_id}", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=FileDeleteResponse, ) 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, ) 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, ) 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, ) 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, )