89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
|
|
import json
|
|
import requests
|
|
import urllib
|
|
|
|
|
|
class OpenAPIMethod:
|
|
def __init__(self, url, name, descriptor, catalog):
|
|
self.url = url
|
|
self.__name__ = name
|
|
|
|
assert 'post' in descriptor, 'Only POST methods are supported'
|
|
post_descriptor = descriptor['post']
|
|
|
|
self.__doc__ = post_descriptor['description']
|
|
parameters = post_descriptor.get('parameters', [])
|
|
request_body = post_descriptor.get('requestBody')
|
|
|
|
self.parameters = {p['name']: p for p in parameters}
|
|
assert all(param['in'] == 'query' for param in self.parameters.values()), f'Only query path parameters are supported (path: {path}, descriptor: {json.dumps(descriptor)})'
|
|
|
|
self.body = None
|
|
self.body_name = None
|
|
if request_body:
|
|
assert 'application/json' in request_body['content'], f'Only application/json is supported for request body (path: {path}, descriptor: {json.dumps(descriptor)})'
|
|
self.body = dict(
|
|
required=request_body['required'],
|
|
schema=request_body['content']['application/json']['schema'],
|
|
)
|
|
|
|
self.body_name = 'body'
|
|
i = 2
|
|
while self.body_name in self.parameters:
|
|
self.body_name = f'body{i}'
|
|
i += 1
|
|
|
|
self.parameters_schema = dict(
|
|
type='object',
|
|
properties={
|
|
**({
|
|
self.body_name: self.body['schema']
|
|
} if self.body else {}),
|
|
**{
|
|
name: param['schema']
|
|
for name, param in self.parameters.items()
|
|
}
|
|
},
|
|
components=catalog.get('components'),
|
|
required=[name for name, param in self.parameters.items() if param['required']] + ([self.body_name] if self.body and self.body['required'] else [])
|
|
)
|
|
|
|
def __call__(self, **kwargs):
|
|
if self.body:
|
|
body = kwargs.pop(self.body_name, None)
|
|
if self.body['required']:
|
|
assert body is not None, f'Missing required body parameter: {self.body_name}'
|
|
else:
|
|
body = None
|
|
|
|
query_params = {}
|
|
for name, param in self.parameters.items():
|
|
value = kwargs.pop(name, None)
|
|
if param['required']:
|
|
assert value is not None, f'Missing required parameter: {name}'
|
|
|
|
assert param['in'] == 'query', 'Only query parameters are supported'
|
|
query_params[name] = value
|
|
|
|
params = "&".join(f"{name}={urllib.parse.quote(value)}" for name, value in query_params.items())
|
|
url = f'{self.url}?{params}'
|
|
response = requests.post(url, json=body)
|
|
response.raise_for_status()
|
|
response_json = response.json()
|
|
|
|
return response_json
|
|
|
|
|
|
def openapi_methods_from_endpoint(url):
|
|
catalog_url = f'{url}/openapi.json'
|
|
catalog_response = requests.get(catalog_url)
|
|
catalog_response.raise_for_status()
|
|
catalog = catalog_response.json()
|
|
|
|
methods = [
|
|
OpenAPIMethod(url=f'{url}{path}', name=path.replace('/', ' ').strip().replace(' ', '_'), descriptor=descriptor, catalog=catalog)
|
|
for path, descriptor in catalog['paths'].items()
|
|
]
|
|
return methods
|