32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import json
|
|
from os import path
|
|
from typing import Tuple, Type, Union, List
|
|
|
|
from openapi_core.validation.request.datatypes import OpenAPIRequest
|
|
|
|
from cic_eth.server.openapi.UWSGIOpenAPIRequest import UWSGIOpenAPIRequest
|
|
from openapi_core import create_spec
|
|
from openapi_core.validation.request.validators import RequestValidator
|
|
from openapi_spec_validator.schemas import read_yaml_file
|
|
|
|
spec_dict = read_yaml_file(path.join(path.dirname(
|
|
__file__), './server.yaml'))
|
|
spec = create_spec(spec_dict)
|
|
|
|
|
|
def request(env, start_response) -> Tuple[Union[List[bytes], None], Type[OpenAPIRequest]]:
|
|
oAPIRequest = UWSGIOpenAPIRequest(env)
|
|
validator = RequestValidator(spec)
|
|
result = validator.validate(oAPIRequest)
|
|
|
|
if result.errors:
|
|
json_data = json.dumps(list(map(lambda e: str(e), result.errors)))
|
|
content = json_data.encode('utf-8')
|
|
headers = []
|
|
headers.append(('Content-Length', str(len(content))),)
|
|
headers.append(('Access-Control-Allow-Origin', '*',))
|
|
headers.append(('Content-Type', 'application/json',))
|
|
start_response('400 Invalid Request', headers)
|
|
return ([content], oAPIRequest)
|
|
return (None, oAPIRequest)
|