openai/openai-python
Publicmirrored from https://github.com/openai/openai-pythonAvailable
openai/api_resources/abstract/nested_resource_class_methods.py
102lines · modecode
| 1 | from urllib.parse import quote_plus |
| 2 | |
| 3 | from openai import api_requestor, util |
| 4 | |
| 5 | |
| 6 | def nested_resource_class_methods( |
| 7 | resource, path=None, operations=None, resource_plural=None |
| 8 | ): |
| 9 | if resource_plural is None: |
| 10 | resource_plural = "%ss" % resource |
| 11 | if path is None: |
| 12 | path = resource_plural |
| 13 | if operations is None: |
| 14 | raise ValueError("operations list required") |
| 15 | |
| 16 | def wrapper(cls): |
| 17 | def nested_resource_url(cls, id, nested_id=None): |
| 18 | url = "%s/%s/%s" % (cls.class_url(), quote_plus(id), quote_plus(path)) |
| 19 | if nested_id is not None: |
| 20 | url += "/%s" % quote_plus(nested_id) |
| 21 | return url |
| 22 | |
| 23 | resource_url_method = "%ss_url" % resource |
| 24 | setattr(cls, resource_url_method, classmethod(nested_resource_url)) |
| 25 | |
| 26 | def nested_resource_request( |
| 27 | cls, |
| 28 | method, |
| 29 | url, |
| 30 | api_key=None, |
| 31 | request_id=None, |
| 32 | api_version=None, |
| 33 | organization=None, |
| 34 | **params, |
| 35 | ): |
| 36 | requestor = api_requestor.APIRequestor( |
| 37 | api_key, api_version=api_version, organization=organization |
| 38 | ) |
| 39 | response, _, api_key = requestor.request( |
| 40 | method, url, params, request_id=request_id |
| 41 | ) |
| 42 | return util.convert_to_openai_object( |
| 43 | response, api_key, api_version, organization |
| 44 | ) |
| 45 | |
| 46 | resource_request_method = "%ss_request" % resource |
| 47 | setattr(cls, resource_request_method, classmethod(nested_resource_request)) |
| 48 | |
| 49 | for operation in operations: |
| 50 | if operation == "create": |
| 51 | |
| 52 | def create_nested_resource(cls, id, **params): |
| 53 | url = getattr(cls, resource_url_method)(id) |
| 54 | return getattr(cls, resource_request_method)("post", url, **params) |
| 55 | |
| 56 | create_method = "create_%s" % resource |
| 57 | setattr(cls, create_method, classmethod(create_nested_resource)) |
| 58 | |
| 59 | elif operation == "retrieve": |
| 60 | |
| 61 | def retrieve_nested_resource(cls, id, nested_id, **params): |
| 62 | url = getattr(cls, resource_url_method)(id, nested_id) |
| 63 | return getattr(cls, resource_request_method)("get", url, **params) |
| 64 | |
| 65 | retrieve_method = "retrieve_%s" % resource |
| 66 | setattr(cls, retrieve_method, classmethod(retrieve_nested_resource)) |
| 67 | |
| 68 | elif operation == "update": |
| 69 | |
| 70 | def modify_nested_resource(cls, id, nested_id, **params): |
| 71 | url = getattr(cls, resource_url_method)(id, nested_id) |
| 72 | return getattr(cls, resource_request_method)("post", url, **params) |
| 73 | |
| 74 | modify_method = "modify_%s" % resource |
| 75 | setattr(cls, modify_method, classmethod(modify_nested_resource)) |
| 76 | |
| 77 | elif operation == "delete": |
| 78 | |
| 79 | def delete_nested_resource(cls, id, nested_id, **params): |
| 80 | url = getattr(cls, resource_url_method)(id, nested_id) |
| 81 | return getattr(cls, resource_request_method)( |
| 82 | "delete", url, **params |
| 83 | ) |
| 84 | |
| 85 | delete_method = "delete_%s" % resource |
| 86 | setattr(cls, delete_method, classmethod(delete_nested_resource)) |
| 87 | |
| 88 | elif operation == "list": |
| 89 | |
| 90 | def list_nested_resources(cls, id, **params): |
| 91 | url = getattr(cls, resource_url_method)(id) |
| 92 | return getattr(cls, resource_request_method)("get", url, **params) |
| 93 | |
| 94 | list_method = "list_%s" % resource_plural |
| 95 | setattr(cls, list_method, classmethod(list_nested_resources)) |
| 96 | |
| 97 | else: |
| 98 | raise ValueError("Unknown operation: %s" % operation) |
| 99 | |
| 100 | return cls |
| 101 | |
| 102 | return wrapper |
| 103 | |