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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
|
from __future__ import annotations
import warnings
from typing import TYPE_CHECKING, Any, AsyncGenerator, Generic
from litestar._multipart import parse_content_header, parse_multipart_form
from litestar._parsers import parse_url_encoded_form_data
from litestar.connection.base import (
ASGIConnection,
AuthT,
StateT,
UserT,
empty_receive,
empty_send,
)
from litestar.datastructures.headers import Accept
from litestar.datastructures.multi_dicts import FormMultiDict
from litestar.enums import ASGIExtension, RequestEncodingType
from litestar.exceptions import (
InternalServerException,
LitestarException,
LitestarWarning,
)
from litestar.serialization import decode_json, decode_msgpack
from litestar.types import Empty
__all__ = ("Request",)
if TYPE_CHECKING:
from litestar.handlers.http_handlers import HTTPRouteHandler # noqa: F401
from litestar.types.asgi_types import HTTPScope, Method, Receive, Scope, Send
from litestar.types.empty import EmptyType
SERVER_PUSH_HEADERS = {
"accept",
"accept-encoding",
"accept-language",
"cache-control",
"user-agent",
}
class Request(Generic[UserT, AuthT, StateT], ASGIConnection["HTTPRouteHandler", UserT, AuthT, StateT]):
"""The Litestar Request class."""
__slots__ = (
"_json",
"_form",
"_body",
"_msgpack",
"_content_type",
"_accept",
"is_connected",
"supports_push_promise",
)
scope: HTTPScope # pyright: ignore
"""The ASGI scope attached to the connection."""
receive: Receive
"""The ASGI receive function."""
send: Send
"""The ASGI send function."""
def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send) -> None:
"""Initialize ``Request``.
Args:
scope: The ASGI connection scope.
receive: The ASGI receive function.
send: The ASGI send function.
"""
super().__init__(scope, receive, send)
self.is_connected: bool = True
self._body: bytes | EmptyType = Empty
self._form: dict[str, str | list[str]] | EmptyType = Empty
self._json: Any = Empty
self._msgpack: Any = Empty
self._content_type: tuple[str, dict[str, str]] | EmptyType = Empty
self._accept: Accept | EmptyType = Empty
self.supports_push_promise = ASGIExtension.SERVER_PUSH in self._server_extensions
@property
def method(self) -> Method:
"""Return the request method.
Returns:
The request :class:`Method <litestar.types.Method>`
"""
return self.scope["method"]
@property
def content_type(self) -> tuple[str, dict[str, str]]:
"""Parse the request's 'Content-Type' header, returning the header value and any options as a dictionary.
Returns:
A tuple with the parsed value and a dictionary containing any options send in it.
"""
if self._content_type is Empty:
if (content_type := self._connection_state.content_type) is not Empty:
self._content_type = content_type
else:
self._content_type = self._connection_state.content_type = parse_content_header(
self.headers.get("Content-Type", "")
)
return self._content_type
@property
def accept(self) -> Accept:
"""Parse the request's 'Accept' header, returning an :class:`Accept <litestar.datastructures.headers.Accept>` instance.
Returns:
An :class:`Accept <litestar.datastructures.headers.Accept>` instance, representing the list of acceptable media types.
"""
if self._accept is Empty:
if (accept := self._connection_state.accept) is not Empty:
self._accept = accept
else:
self._accept = self._connection_state.accept = Accept(self.headers.get("Accept", "*/*"))
return self._accept
async def json(self) -> Any:
"""Retrieve the json request body from the request.
Returns:
An arbitrary value
"""
if self._json is Empty:
if (json_ := self._connection_state.json) is not Empty:
self._json = json_
else:
body = await self.body()
self._json = self._connection_state.json = decode_json(
body or b"null", type_decoders=self.route_handler.resolve_type_decoders()
)
return self._json
async def msgpack(self) -> Any:
"""Retrieve the MessagePack request body from the request.
Returns:
An arbitrary value
"""
if self._msgpack is Empty:
if (msgpack := self._connection_state.msgpack) is not Empty:
self._msgpack = msgpack
else:
body = await self.body()
self._msgpack = self._connection_state.msgpack = decode_msgpack(
body or b"\xc0", type_decoders=self.route_handler.resolve_type_decoders()
)
return self._msgpack
async def stream(self) -> AsyncGenerator[bytes, None]:
"""Return an async generator that streams chunks of bytes.
Returns:
An async generator.
Raises:
RuntimeError: if the stream is already consumed
"""
if self._body is Empty:
if not self.is_connected:
raise InternalServerException("stream consumed")
while event := await self.receive():
if event["type"] == "http.request":
if event["body"]:
yield event["body"]
if not event.get("more_body", False):
break
if event["type"] == "http.disconnect":
raise InternalServerException("client disconnected prematurely")
self.is_connected = False
yield b""
else:
yield self._body
yield b""
return
async def body(self) -> bytes:
"""Return the body of the request.
Returns:
A byte-string representing the body of the request.
"""
if self._body is Empty:
if (body := self._connection_state.body) is not Empty:
self._body = body
else:
self._body = self._connection_state.body = b"".join([c async for c in self.stream()])
return self._body
async def form(self) -> FormMultiDict:
"""Retrieve form data from the request. If the request is either a 'multipart/form-data' or an
'application/x-www-form- urlencoded', return a FormMultiDict instance populated with the values sent in the
request, otherwise, an empty instance.
Returns:
A FormMultiDict instance
"""
if self._form is Empty:
if (form := self._connection_state.form) is not Empty:
self._form = form
else:
content_type, options = self.content_type
if content_type == RequestEncodingType.MULTI_PART:
self._form = parse_multipart_form(
body=await self.body(),
boundary=options.get("boundary", "").encode(),
multipart_form_part_limit=self.app.multipart_form_part_limit,
)
elif content_type == RequestEncodingType.URL_ENCODED:
self._form = parse_url_encoded_form_data(
await self.body(),
)
else:
self._form = {}
self._connection_state.form = self._form
return FormMultiDict(self._form)
async def send_push_promise(self, path: str, raise_if_unavailable: bool = False) -> None:
"""Send a push promise.
This method requires the `http.response.push` extension to be sent from the ASGI server.
Args:
path: Path to send the promise to.
raise_if_unavailable: Raise an exception if server push is not supported by
the server
Returns:
None
"""
if not self.supports_push_promise:
if raise_if_unavailable:
raise LitestarException("Attempted to send a push promise but the server does not support it")
warnings.warn(
"Attempted to send a push promise but the server does not support it. In a future version, this will "
"raise an exception. To enable this behaviour in the current version, set raise_if_unavailable=True. "
"To prevent this behaviour, make sure that the server you are using supports the 'http.response.push' "
"ASGI extension, or check this dynamically via "
":attr:`~litestar.connection.Request.supports_push_promise`",
stacklevel=2,
category=LitestarWarning,
)
return
raw_headers = [
(header_name.encode("latin-1"), value.encode("latin-1"))
for header_name in (self.headers.keys() & SERVER_PUSH_HEADERS)
for value in self.headers.getall(header_name, [])
]
await self.send({"type": "http.response.push", "path": path, "headers": raw_headers})
|
