1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
"""OpenAPI core contrib requests requests module"""
from typing import Optional
from typing import Union
from urllib.parse import parse_qs
from urllib.parse import urlparse
from requests import PreparedRequest
from requests import Request
from requests.cookies import RequestsCookieJar
from werkzeug.datastructures import Headers
from werkzeug.datastructures import ImmutableMultiDict
from openapi_core.contrib.requests.protocols import SupportsCookieJar
from openapi_core.datatypes import RequestParameters
class RequestsOpenAPIRequest:
"""
Converts a requests request to an OpenAPI request
Internally converts to a `PreparedRequest` first to parse the exact
payload being sent
"""
def __init__(self, request: Union[Request, PreparedRequest]):
if not isinstance(request, (Request, PreparedRequest)):
raise TypeError(
"'request' argument is not type of "
f"{Request} or {PreparedRequest}"
)
if isinstance(request, Request):
request = request.prepare()
self.request = request
if request.url is None:
raise RuntimeError("Request URL is missing")
self._url_parsed = urlparse(request.url, allow_fragments=False)
cookie = {}
if isinstance(self.request, SupportsCookieJar) and isinstance(
self.request._cookies, RequestsCookieJar
):
# cookies are stored in a cookiejar object
cookie = self.request._cookies.get_dict()
self.parameters = RequestParameters(
query=ImmutableMultiDict(parse_qs(self._url_parsed.query)),
header=Headers(dict(self.request.headers)),
cookie=ImmutableMultiDict(cookie),
)
@property
def host_url(self) -> str:
return f"{self._url_parsed.scheme}://{self._url_parsed.netloc}"
@property
def path(self) -> str:
assert isinstance(self._url_parsed.path, str)
return self._url_parsed.path
@property
def method(self) -> str:
method = self.request.method
return method and method.lower() or ""
@property
def body(self) -> Optional[bytes]:
if self.request.body is None:
return None
if isinstance(self.request.body, bytes):
return self.request.body
assert isinstance(self.request.body, str)
# TODO: figure out if request._body_position is relevant
return self.request.body.encode("utf-8")
@property
def content_type(self) -> str:
# Order matters because all python requests issued from a session
# include Accept */* which does not necessarily match the content type
return str(
self.request.headers.get("Content-Type")
or self.request.headers.get("Accept")
)
class RequestsOpenAPIWebhookRequest(RequestsOpenAPIRequest):
"""
Converts a requests request to an OpenAPI Webhook request
Internally converts to a `PreparedRequest` first to parse the exact
payload being sent
"""
def __init__(self, request: Union[Request, PreparedRequest], name: str):
super().__init__(request)
self.name = name
|
