openai/openai-python
Publicmirrored from https://github.com/openai/openai-pythonAvailable
openai/api_resources/file.py
131lines · modecode
| 1 | import json |
| 2 | import os |
| 3 | from typing import cast |
| 4 | |
| 5 | import openai |
| 6 | from openai import api_requestor, util, error |
| 7 | from openai.api_resources.abstract import DeletableAPIResource, ListableAPIResource |
| 8 | from openai.util import ApiType |
| 9 | |
| 10 | |
| 11 | class File(ListableAPIResource, DeletableAPIResource): |
| 12 | OBJECT_NAME = "files" |
| 13 | |
| 14 | @classmethod |
| 15 | def create( |
| 16 | cls, |
| 17 | file, |
| 18 | purpose, |
| 19 | model=None, |
| 20 | api_key=None, |
| 21 | api_base=None, |
| 22 | api_type=None, |
| 23 | api_version=None, |
| 24 | organization=None, |
| 25 | user_provided_filename=None, |
| 26 | ): |
| 27 | if purpose != "search" and model is not None: |
| 28 | raise ValueError("'model' is only meaningful if 'purpose' is 'search'") |
| 29 | requestor = api_requestor.APIRequestor( |
| 30 | api_key, |
| 31 | api_base=api_base or openai.api_base, |
| 32 | api_type=api_type, |
| 33 | api_version=api_version, |
| 34 | organization=organization, |
| 35 | ) |
| 36 | typed_api_type, api_version = cls._get_api_type_and_version(api_type, api_version) |
| 37 | |
| 38 | if typed_api_type == ApiType.AZURE: |
| 39 | base = cls.class_url() |
| 40 | url = "/%s%s?api-version=%s" % (cls.azure_api_prefix, base, api_version) |
| 41 | elif typed_api_type == ApiType.OPEN_AI: |
| 42 | url = cls.class_url() |
| 43 | else: |
| 44 | raise error.InvalidAPIType('Unsupported API type %s' % api_type) |
| 45 | |
| 46 | # Set the filename on 'purpose' and 'model' to None so they are |
| 47 | # interpreted as form data. |
| 48 | files = [("purpose", (None, purpose))] |
| 49 | if model is not None: |
| 50 | files.append(("model", (None, model))) |
| 51 | if user_provided_filename is not None: |
| 52 | files.append(("file", (user_provided_filename, file, 'application/octet-stream'))) |
| 53 | else: |
| 54 | files.append(("file", ("file", file, 'application/octet-stream'))) |
| 55 | response, _, api_key = requestor.request("post", url, files=files) |
| 56 | return util.convert_to_openai_object( |
| 57 | response, api_key, api_version, organization |
| 58 | ) |
| 59 | |
| 60 | @classmethod |
| 61 | def download( |
| 62 | cls, |
| 63 | id, |
| 64 | api_key=None, |
| 65 | api_base=None, |
| 66 | api_type=None, |
| 67 | api_version=None, |
| 68 | organization=None |
| 69 | ): |
| 70 | requestor = api_requestor.APIRequestor( |
| 71 | api_key, |
| 72 | api_base=api_base or openai.api_base, |
| 73 | api_type=api_type, |
| 74 | api_version=api_version, |
| 75 | organization=organization, |
| 76 | ) |
| 77 | typed_api_type, api_version = cls._get_api_type_and_version(api_type, api_version) |
| 78 | |
| 79 | if typed_api_type == ApiType.AZURE: |
| 80 | base = cls.class_url() |
| 81 | url = "/%s%s/%s/content?api-version=%s" % (cls.azure_api_prefix, base, id, api_version) |
| 82 | elif typed_api_type == ApiType.OPEN_AI: |
| 83 | url = f"{cls.class_url()}/{id}/content" |
| 84 | else: |
| 85 | raise error.InvalidAPIType('Unsupported API type %s' % api_type) |
| 86 | |
| 87 | result = requestor.request_raw("get", url) |
| 88 | if not 200 <= result.status_code < 300: |
| 89 | raise requestor.handle_error_response( |
| 90 | result.content, |
| 91 | result.status_code, |
| 92 | json.loads(cast(bytes, result.content)), |
| 93 | result.headers, |
| 94 | stream_error=False, |
| 95 | ) |
| 96 | return result.content |
| 97 | |
| 98 | @classmethod |
| 99 | def find_matching_files( |
| 100 | cls, |
| 101 | name, |
| 102 | bytes, |
| 103 | purpose, |
| 104 | api_key=None, |
| 105 | api_base=None, |
| 106 | api_type=None, |
| 107 | api_version=None, |
| 108 | organization=None, |
| 109 | ): |
| 110 | """Find already uploaded files with the same name, size, and purpose.""" |
| 111 | all_files = cls.list( |
| 112 | api_key=api_key, |
| 113 | api_base=api_base or openai.api_base, |
| 114 | api_type=api_type, |
| 115 | api_version=api_version, |
| 116 | organization=organization, |
| 117 | ).get("data", []) |
| 118 | matching_files = [] |
| 119 | basename = os.path.basename(name) |
| 120 | for f in all_files: |
| 121 | if f["purpose"] != purpose: |
| 122 | continue |
| 123 | file_basename = os.path.basename(f["filename"]) |
| 124 | if file_basename != basename: |
| 125 | continue |
| 126 | if "bytes" in f and f["bytes"] != bytes: |
| 127 | continue |
| 128 | if "size" in f and int(f["size"]) != bytes: |
| 129 | continue |
| 130 | matching_files.append(f) |
| 131 | return matching_files |
| 132 | |