58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""OpenAPI core contrib requests requests module"""
|
|
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from openapi_core.validation.request.datatypes import (OpenAPIRequest,
|
|
RequestParameters)
|
|
from requests import Request
|
|
from werkzeug.datastructures import Headers, ImmutableMultiDict
|
|
from werkzeug.wrappers import Request
|
|
|
|
class OpenAPIRequestFactory:
|
|
@classmethod
|
|
def create(cls, env):
|
|
"""
|
|
Converts a requests request to an OpenAPI one
|
|
|
|
"""
|
|
request = Request(env)
|
|
|
|
# Method
|
|
method = request.method.lower()
|
|
|
|
# Preparing a request formats the URL with params, strip them out again
|
|
o = urlparse(request.url)
|
|
params = parse_qs(o.query)
|
|
# extract the URL without query parameters
|
|
url = o._replace(query=None).geturl()
|
|
|
|
# Order matters because all python requests issued from a session
|
|
# include Accept */* which does not necessarily match the content type
|
|
mimetype = request.headers.get("Content-Type") or request.headers.get(
|
|
"Accept"
|
|
)
|
|
|
|
# Headers - request.headers is not an instance of Headers
|
|
# which is expected
|
|
header = Headers(dict(request.headers))
|
|
|
|
# Body
|
|
body = request.get_data()
|
|
|
|
# Path gets deduced by path finder against spec
|
|
parameters = RequestParameters(
|
|
query=ImmutableMultiDict(params),
|
|
header=header,
|
|
cookie=request.cookies,
|
|
)
|
|
return OpenAPIRequest(
|
|
full_url_pattern=url,
|
|
method=method,
|
|
parameters=parameters,
|
|
body=body,
|
|
mimetype=mimetype,
|
|
)
|
|
|
|
|
|
UWSGIOpenAPIRequest = OpenAPIRequestFactory.create
|