跳到内容

Request 类

你可以在路径操作函数或依赖项中声明一个类型为 Request 的参数，然后就可以直接访问原始请求对象，而无需进行任何验证等操作。

在 FastAPI 关于直接使用 Request 的文档中了解更多相关信息

您可以直接从 fastapi 导入它。

from fastapi import Request

提示

当您想定义应同时兼容 HTTP 和 WebSocket 的依赖项时，可以定义一个参数，该参数接受 HTTPConnection 而不是 Request 或 WebSocket。

fastapi.Request 

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

基类： HTTPConnection" href="../httpconnection/#fastapi.requests.HTTPConnection">HTTPConnection[StateT]

源代码在 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 实例属性 

scope = scope

app 属性 

app

url 属性 

url

base_url 属性 

base_url

headers 属性 

headers

query_params 属性 

query_params

path_params 属性 

path_params

cookies 属性 

cookies

client 属性 

client

session 属性 

session

auth 属性 

auth

user 属性 

user

state 属性 

state

method 属性 

method

receive 属性 

receive

url_for 

url_for(name, /, **path_params)
源代码在 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 异步 

stream()
源代码在 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 异步 

body()
源代码在 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 异步 

json()
源代码在 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
)
源代码在 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 异步 

close()
源代码在 starlette/requests.py 
async def close(self) -> None:
    if self._form is not None:  # pragma: no branch
        await self._form.close()

is_disconnected 异步 

is_disconnected()
源代码在 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 异步 

send_push_promise(path)
源代码在 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})