Request 類別

您可以在*路徑操作函式*或依賴項中宣告一個類型為 Request 的參數，然後您可以直接存取原始請求物件，無需任何驗證等等。

您可以直接從 fastapi 導入它

from fastapi import Request

提示

當您想要定義與 HTTP 和 WebSocket 相容的依賴項時，您可以定義一個接受 HTTPConnection 而不是 Request 或 WebSocket 的參數。

fastapi.Request

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

基底： HTTPConnection

參數	說明
scope	類型： Scope
receive	類型： Receive 預設值： empty_receive
send	類型： Send 預設值： empty_send

原始碼位於 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 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)

參數	說明
名稱	類型： str
**path_params	類型： Any 預設值： {}

原始碼位於 starlette/requests.py

def url_for(self, name: str, /, **path_params: typing.Any) -> URL:
    router: Router = self.scope["router"]
    url_path = router.url_path_for(name, **path_params)
    return url_path.make_absolute_url(base_url=self.base_url)

stream async

stream()

原始碼位於 starlette/requests.py

async def stream(self) -> typing.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":
            self._is_disconnected = True
            raise ClientDisconnect()
    yield b""

body async

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 async

json()

原始碼位於 starlette/requests.py

async def json(self) -> typing.Any:
    if not hasattr(self, "_json"):
        body = await self.body()
        self._json = json.loads(body)
    return self._json

form

form(*, max_files=1000, max_fields=1000)

參數	說明
max_files（最大檔案數）	類型： int | float 預設值： 1000
max_fields（最大欄位數）	類型： int | float 預設值： 1000

原始碼位於 starlette/requests.py

def form(
    self, *, max_files: int | float = 1000, max_fields: int | float = 1000
) -> AwaitableOrContextManager[FormData]:
    return AwaitableOrContextManagerWrapper(self._get_form(max_files=max_files, max_fields=max_fields))

關閉 async

close()

原始碼位於 starlette/requests.py

async def close(self) -> None:
    if self._form is not None:
        await self._form.close()

已斷線 async

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

發送推播承諾 async

send_push_promise(path)

參數	說明
路徑	類型： str

原始碼位於 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})