import json import os from typing import cast import time import openai from openai import api_requestor, util, error from openai.api_resources.abstract import DeletableAPIResource, ListableAPIResource from openai.util import ApiType class File(ListableAPIResource, DeletableAPIResource): OBJECT_NAME = "files" @classmethod def __prepare_file_create( cls, file, purpose, model=None, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, user_provided_filename=None, ): requestor = api_requestor.APIRequestor( api_key, api_base=api_base or openai.api_base, api_type=api_type, api_version=api_version, organization=organization, ) typed_api_type, api_version = cls._get_api_type_and_version( api_type, api_version ) if typed_api_type in (ApiType.AZURE, ApiType.AZURE_AD): base = cls.class_url() url = "/%s%s?api-version=%s" % (cls.azure_api_prefix, base, api_version) elif typed_api_type == ApiType.OPEN_AI: url = cls.class_url() else: raise error.InvalidAPIType("Unsupported API type %s" % api_type) # Set the filename on 'purpose' and 'model' to None so they are # interpreted as form data. files = [("purpose", (None, purpose))] if model is not None: files.append(("model", (None, model))) if user_provided_filename is not None: files.append( ("file", (user_provided_filename, file, "application/octet-stream")) ) else: files.append(("file", ("file", file, "application/octet-stream"))) return requestor, url, files @classmethod def create( cls, file, purpose, model=None, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, user_provided_filename=None, ): requestor, url, files = cls.__prepare_file_create( file, purpose, model, api_key, api_base, api_type, api_version, organization, user_provided_filename, ) response, _, api_key = requestor.request("post", url, files=files) return util.convert_to_openai_object( response, api_key, api_version, organization ) @classmethod async def acreate( cls, file, purpose, model=None, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, user_provided_filename=None, ): requestor, url, files = cls.__prepare_file_create( file, purpose, model, api_key, api_base, api_type, api_version, organization, user_provided_filename, ) response, _, api_key = await requestor.arequest("post", url, files=files) return util.convert_to_openai_object( response, api_key, api_version, organization ) @classmethod def __prepare_file_download( cls, id, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, ): requestor = api_requestor.APIRequestor( api_key, api_base=api_base or openai.api_base, api_type=api_type, api_version=api_version, organization=organization, ) typed_api_type, api_version = cls._get_api_type_and_version( api_type, api_version ) if typed_api_type in (ApiType.AZURE, ApiType.AZURE_AD): base = cls.class_url() url = f"/{cls.azure_api_prefix}{base}/{id}/content?api-version={api_version}" elif typed_api_type == ApiType.OPEN_AI: url = f"{cls.class_url()}/{id}/content" else: raise error.InvalidAPIType("Unsupported API type %s" % api_type) return requestor, url @classmethod def download( cls, id, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, ): requestor, url = cls.__prepare_file_download( id, api_key, api_base, api_type, api_version, organization ) result = requestor.request_raw("get", url) if not 200 <= result.status_code < 300: raise requestor.handle_error_response( result.content, result.status_code, json.loads(cast(bytes, result.content)), result.headers, stream_error=False, ) return result.content @classmethod async def adownload( cls, id, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, ): requestor, url = cls.__prepare_file_download( id, api_key, api_base, api_type, api_version, organization ) async with api_requestor.aiohttp_session() as session: result = await requestor.arequest_raw("get", url, session) if not 200 <= result.status < 300: raise requestor.handle_error_response( result.content, result.status, json.loads(cast(bytes, result.content)), result.headers, stream_error=False, ) return result.content @classmethod def __find_matching_files(cls, name, bytes, all_files, purpose): matching_files = [] basename = os.path.basename(name) for f in all_files: if f["purpose"] != purpose: continue file_basename = os.path.basename(f["filename"]) if file_basename != basename: continue if "bytes" in f and f["bytes"] != bytes: continue if "size" in f and int(f["size"]) != bytes: continue matching_files.append(f) return matching_files @classmethod def find_matching_files( cls, name, bytes, purpose, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, ): """Find already uploaded files with the same name, size, and purpose.""" all_files = cls.list( api_key=api_key, api_base=api_base or openai.api_base, api_type=api_type, api_version=api_version, organization=organization, ).get("data", []) return cls.__find_matching_files(name, bytes, all_files, purpose) @classmethod async def afind_matching_files( cls, name, bytes, purpose, api_key=None, api_base=None, api_type=None, api_version=None, organization=None, ): """Find already uploaded files with the same name, size, and purpose.""" all_files = ( await cls.alist( api_key=api_key, api_base=api_base or openai.api_base, api_type=api_type, api_version=api_version, organization=organization, ) ).get("data", []) return cls.__find_matching_files(name, bytes, all_files, purpose) @classmethod def wait_for_processing(cls, id, max_wait_seconds=30 * 60): TERMINAL_STATES = ["processed", "error", "deleted"] start = time.time() file = cls.retrieve(id=id) while file.status not in TERMINAL_STATES: file = cls.retrieve(id=id) time.sleep(5.0) if time.time() - start > max_wait_seconds: raise openai.error.OpenAIError( message="Giving up on waiting for file {id} to finish processing after {max_wait_seconds} seconds.".format( id=id, max_wait_seconds=max_wait_seconds ) ) return file.status