Saltar al contenido

Clase Request

Puedes declarar un parámetro en una función de path operation o dependencia de tipo Request y luego puedes acceder al objeto request directamente, sin validación, etc.

Lee más sobre esto en la documentación de FastAPI sobre usar Request directamente

Puedes importarlo directamente desde fastapi:

from fastapi import Request

Consejo

Cuando quieres definir dependencias que deberían ser compatibles tanto con HTTP como con WebSockets, puedes definir un parámetro que tome un HTTPConnection en lugar de un Request o un WebSocket.

fastapi.Request 

Request(scope, receive=empty_receive, send=empty_send)

Bases: HTTPConnection[StateT]

Código fuente en 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 = None

scope instance-attribute 

scope = scope

app property 

app

url property 

url

base_url property 

base_url

headers property 

headers

query_params property 

query_params

path_params property 

path_params

cookies property 

cookies

client property 

client

session property 

session

auth property 

auth

user property 

user

state property 

state

method property 

method

receive property 

receive

url_for 

url_for(name, /, **path_params)
Código fuente en 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 

stream()
Código fuente en starlette/requests.py 
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 

body()
Código fuente en starlette/requests.py 
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 

json()
Código fuente en 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 

form(
    *,
    max_files=1000,
    max_fields=1000,
    max_part_size=1024 * 1024
)
Código fuente en starlette/requests.py 
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 

close()
Código fuente en starlette/requests.py 
async def close(self) -> None:
    if self._form is not None:  # pragma: no branch
        await self._form.close()

is_disconnected async 

is_disconnected()
Código fuente en starlette/requests.py 
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 

send_push_promise(path)
Código fuente en starlette/requests.py 
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})