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.
Read more about it in the FastAPI docs about using Request directly
You can import it directly from fastapi:
from fastapi import Requestfastapi.Request ¶
Section titled “fastapi.Request ¶”Request(scope, receive=empty_receive, send=empty_send)Bases: HTTPConnection[StateT]
Source code in starlette/requests.py
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 = Nonescope instance-attribute ¶
Section titled “scope instance-attribute ¶”scope = scopeapp property ¶
Section titled “app property ¶”appurl property ¶
Section titled “url property ¶”urlbase_url property ¶
Section titled “base_url property ¶”base_urlheaders property ¶
Section titled “headers property ¶”headersquery_params property ¶
Section titled “query_params property ¶”query_paramspath_params property ¶
Section titled “path_params property ¶”path_paramscookies property ¶
Section titled “cookies property ¶”cookiesclient property ¶
Section titled “client property ¶”clientsession property ¶
Section titled “session property ¶”sessionauth property ¶
Section titled “auth property ¶”authuser property ¶
Section titled “user property ¶”userstate property ¶
Section titled “state property ¶”statemethod property ¶
Section titled “method property ¶”methodreceive property ¶
Section titled “receive property ¶”receiveurl_for ¶
Section titled “url_for ¶”url_for(name, /, **path_params)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 ¶
Section titled “stream async ¶”stream()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 ¶
Section titled “body async ¶”body()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._bodyjson async ¶
Section titled “json async ¶”json()async def json(self) -> Any:
if not hasattr(self, "_json"): # pragma: no branch
body = await self.body()
self._json = json.loads(body)
return self._jsonform(
*,
max_files=1000,
max_fields=1000,
max_part_size=1024 * 1024
)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 ¶
Section titled “close async ¶”close()async def close(self) -> None:
if self._form is not None: # pragma: no branch
await self._form.close()is_disconnected async ¶
Section titled “is_disconnected async ¶”is_disconnected()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_disconnectedsend_push_promise async ¶
Section titled “send_push_promise async ¶”send_push_promise(path)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})