Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/__init__.py
| Show First 20 Lines • Show All 295 Lines • ▼ Show 20 Lines | def error_handler(exception, encoder): | ||||
| return response | return response | ||||
| class RPCServerApp(Flask): | class RPCServerApp(Flask): | ||||
| """For each endpoint of the given `backend_class`, tells app.route to call | """For each endpoint of the given `backend_class`, tells app.route to call | ||||
| a function that decodes the request and sends it to the backend object | a function that decodes the request and sends it to the backend object | ||||
| provided by the factory. | provided by the factory. | ||||
| :param Any backend_class: The class of the backend, which will be | :param Any backend_class: | ||||
| analyzed to look for API endpoints. | The class of the backend, which will be analyzed to look | ||||
| :param Callable[[], backend_class] backend_factory: A function with no | for API endpoints. | ||||
| argument that returns | :param Optional[Callable[[], backend_class]] backend_factory: | ||||
| an instance of | A function with no argument that returns an instance of | ||||
| `backend_class`.""" | `backend_class`. If unset, defaults to calling `backend_class` | ||||
| constructor directly. | |||||
| """ | |||||
| request_class = BytesRequest | request_class = BytesRequest | ||||
| def __init__(self, *args, backend_class=None, backend_factory=None, | def __init__(self, *args, backend_class=None, backend_factory=None, | ||||
| **kwargs): | **kwargs): | ||||
| super().__init__(*args, **kwargs) | super().__init__(*args, **kwargs) | ||||
| self.backend_class = backend_class | |||||
| if backend_class is not None: | if backend_class is not None: | ||||
| if backend_factory is None: | if backend_factory is None: | ||||
| raise TypeError('Missing argument backend_factory') | backend_factory = backend_class | ||||
| for (meth_name, meth) in backend_class.__dict__.items(): | for (meth_name, meth) in backend_class.__dict__.items(): | ||||
| if hasattr(meth, '_endpoint_path'): | if hasattr(meth, '_endpoint_path'): | ||||
| self.__add_endpoint(meth_name, meth, backend_factory) | self.__add_endpoint(meth_name, meth, backend_factory) | ||||
| def __add_endpoint(self, meth_name, meth, backend_factory): | def __add_endpoint(self, meth_name, meth, backend_factory): | ||||
| from flask import request | from flask import request | ||||
| @self.route('/'+meth._endpoint_path, methods=['POST']) | @self.route('/'+meth._endpoint_path, methods=['POST']) | ||||
| Show All 23 Lines | |||||