Request
class
You can declare a parameter in a path operation function or dependency to be of type Request
and then you can access the raw request object directly, without any validation, etc.
You can import it directly from fastapi
:
from fastapi import Request
Tip
When you want to define dependencies that should be compatible with both HTTP and WebSockets, you can define a parameter that takes an HTTPConnection
instead of a Request
or a WebSocket
.
fastapi.Request
Request(scope, receive=empty_receive, send=empty_send)
Bases: HTTPConnection
Source code in starlette/requests.py
202
203
204
205
206
207
208
209 | def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send):
super().__init__(scope)
assert scope["type"] == "http"
self._receive = receive
self._send = send
self._stream_consumed = False
self._is_disconnected = False
self._form = None
|
url_for
url_for(name, /, **path_params)
Source code in starlette/requests.py
| def url_for(self, name: str, /, **path_params: Any) -> URL:
url_path_provider: Router | Starlette | None = self.scope.get("router") or self.scope.get("app")
if url_path_provider is None:
raise RuntimeError("The `url_for` method can only be used inside a Starlette application or with a router.")
url_path = url_path_provider.url_path_for(name, **path_params)
return url_path.make_absolute_url(base_url=self.base_url)
|
stream
async
Source code in starlette/requests.py
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237 | async def stream(self) -> AsyncGenerator[bytes, None]:
if hasattr(self, "_body"):
yield self._body
yield b""
return
if self._stream_consumed:
raise RuntimeError("Stream consumed")
while not self._stream_consumed:
message = await self._receive()
if message["type"] == "http.request":
body = message.get("body", b"")
if not message.get("more_body", False):
self._stream_consumed = True
if body:
yield body
elif message["type"] == "http.disconnect": # pragma: no branch
self._is_disconnected = True
raise ClientDisconnect()
yield b""
|
body
async
Source code in starlette/requests.py
239
240
241
242
243
244
245 | async def body(self) -> bytes:
if not hasattr(self, "_body"):
chunks: list[bytes] = []
async for chunk in self.stream():
chunks.append(chunk)
self._body = b"".join(chunks)
return self._body
|
json
async
Source code in starlette/requests.py
| async def json(self) -> Any:
if not hasattr(self, "_json"): # pragma: no branch
body = await self.body()
self._json = json.loads(body)
return self._json
|
form(
*,
max_files=1000,
max_fields=1000,
max_part_size=1024 * 1024
)
Source code in starlette/requests.py
288
289
290
291
292
293
294
295
296
297 | def form(
self,
*,
max_files: int | float = 1000,
max_fields: int | float = 1000,
max_part_size: int = 1024 * 1024,
) -> AwaitableOrContextManager[FormData]:
return AwaitableOrContextManagerWrapper(
self._get_form(max_files=max_files, max_fields=max_fields, max_part_size=max_part_size)
)
|
close
async
Source code in starlette/requests.py
| async def close(self) -> None:
if self._form is not None: # pragma: no branch
await self._form.close()
|
is_disconnected
async
Source code in starlette/requests.py
303
304
305
306
307
308
309
310
311
312
313
314
315 | async def is_disconnected(self) -> bool:
if not self._is_disconnected:
message: Message = {}
# If message isn't immediately available, move on
with anyio.CancelScope() as cs:
cs.cancel()
message = await self._receive()
if message.get("type") == "http.disconnect":
self._is_disconnected = True
return self._is_disconnected
|
send_push_promise
async
Source code in starlette/requests.py
317
318
319
320
321
322
323 | async def send_push_promise(self, path: str) -> None:
if "http.response.push" in self.scope.get("extensions", {}):
raw_headers: list[tuple[bytes, bytes]] = []
for name in SERVER_PUSH_HEADERS_TO_COPY:
for value in self.headers.getlist(name):
raw_headers.append((name.encode("latin-1"), value.encode("latin-1")))
await self._send({"type": "http.response.push", "path": path, "headers": raw_headers})
|