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})