如何以编程方式实例化Starlette的主体请求？

Question

我有一个项目，其中包含一些使用FastAPI的API，在该项目中，我需要调用项目中的一个API函数。使用FastAPI和Starlette的API函数如下所示

@router.put("/tab/{tab_name}", tags=["Tab CRUD"])
async def insert_tab(request: Request, tab_name: str):
    tab = await request.json()
    new_tab_name = tab["name"]
    new_tab_title = tab["title"]
    # Rest of the code

我发送了一个包含新选项卡数据的JSON作为请求的主体，稍后将使用await request.json()将其转换为python字典。

现在，我需要在另一个函数中调用insert_tab，因此我需要以某种方式从Starlette实例化Request对象。我以前这样做过，但没有JSON主体：

from starlette.requests import Request
from starlette.datastructures import Headers

headers = Headers()
scope = {
    'method': 'GET',
    'type': 'http',
    'headers': headers
}
request = Request(scope=scope)

但在本例中，我还需要将JSON body注入到Request对象中，但我找不到这样做的方法。

以前有没有人这样做过，或者知道我应该怎么做？

Answer 1

如果您尝试从应用程序中以编程方式调用另一个端点，最好跳过HTTP层，直接调用底层函数。

但是，如果您是在尝试构建一个在单元测试中使用的模拟请求时访问此页面的，下面是一个包含头文件的示例：

from starlette.requests import Request
from starlette.datastructures import Headers
from typing import Dict


def build_request(headers: Dict = None) -> Request:
    if headers is None:
        headers = {}
    return Request({
        "type": "http",
        "headers": Headers(headers).raw
    })

有关使用Starlette的ASGI请求对象的更多信息，请单击此处：

https://www.encode.io/articles/working-with-http-requests-in-asgi

Answer 2

我不是ASGI和HTTP请求方面的专家，但这里有一个让它工作的方法。

要创建一个带有请求体的Request对象，可以传入一个receive参数：

# starlette/requests.py

class Request(HTTPConnection):
    def __init__(
        self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send
    ):

其中receive的类型为Receive

# starlette/types.py

Message = typing.MutableMapping[str, typing.Any]
Receive = typing.Callable[[], typing.Awaitable[Message]]

它是一种支持await的函数，它返回一个Message，这是一种可变映射(比如dict)。receive是一个函数的原因是因为它是预期接收的“"receive" the request body as a stream of messages”通道的一部分

关于传入请求的大部分信息都存储在“作用域”中，并在ASGI应用程序被实例化时呈现。但是，对于请求主体来说，这是不可能的。

为了访问请求正文，我们必须从“接收”通道获取消息流。

您可以从Starlette的请求代码中看到它是如何使用的：

self._stream_consumed = True
while True:
    message = await self._receive()  # <------------- this.
    if message["type"] == "http.request":
        body = message.get("body", b"")
        if body:
            yield body
        if not message.get("more_body", False):
            break
    elif message["type"] == "http.disconnect":
        self._is_disconnected = True
        raise ClientDisconnect()
yield b""

由于您可能已经准备好了请求正文，因此您可以在一个块中伪流它，考虑到"type"、"body" (实际正文/有效负载)的预期关键字，以及设置为False的"more_body"关键字以中断流。

import json
from starlette.requests import Message

async def create_body() -> Message:
    body = {'abc': 123, 'def': {'ghi': 456}}
    return {
        'type': 'http.request',
        'body': json.dumps(body).encode('utf-8'),
        'more_body': False,
    }

然后你可以把它传递给Request的receive=参数来调用你的函数：

from starlette.requests import Request

@router.put('/tab/{tab_name}')
async def insert_tab(request: Request, tab_name: str):
    tab = await request.json()
    return {tab_name: tab}

@router.put('/call-insert-tab')
async def some_other_function():
    req = Request(
        scope={
            'type': 'http',
            'scheme': 'http',
            'method': 'PUT',
            'path': '/tab/abc',
            'raw_path': b'/tab/abc',
            'query_string': b'',
            'headers': {}
        },
        receive=create_body,  # <-------- Pass it here
    )
    return await insert_tab(req, 'abc')

虽然这可能会起作用，但我会说这不是很好。使用Starlette这样的库的目的是不关心这些HTTP请求对象是如何实际创建和处理的。除了示例之外，我没有测试上面的代码，但我觉得它可能会在某个时候崩溃(特别是使用FastAPI的依赖注入)。

因为您用例简单地

发送一个包含新选项卡数据的json作为我的请求的主体，稍后将使用await request.json()将其转换为python字典

为此，您实际上不需要直接使用Request。我建议将request: Request参数替换为tab: Tab模型，如FastAPI教程中传入请求主体：https://fastapi.tiangolo.com/tutorial/body/中所述

from pydantic import BaseModel

class Tab(BaseModel):
    abc: int
    xyz: int

@app.put('/tab/{tab_name}')
async def insert_tab(tab: Tab, tab_name: str):
    print(tab.abc)
    print(tab.xyz)
    return {tab_name: tab}

FastAPI将自动获取底层Request对象的主体，并将其从JSON转换为您的模型。(它自动假设不在路径中的参数是主体/有效负载的一部分。)它的工作方式应该和request.json()一样

$ cat data.json
{
    "abc": 123,
    "xyz": 456
}

$ curl -s -XPUT 'http://localhost:5050/tab/tab1' --header 'Content-Type: application/json' --data @data.json | jq
{
  "tab1": {
    "abc": 123,
    "xyz": 456
  }
}

这也使得调用需要请求体的路由函数变得更容易:您只需直接传入数据，而无需使用Requests

@app.put('/tab/{tab_name}')
async def insert_tab(tab: Tab, tab_name: str):
    return {tab_name: tab}

@app.put('/call-insert-tab')
async def some_other_function():
    tab = Tab(abc=123, xyz=456)
    return await insert_tab(tab, 'tab1')

有关更多信息和示例，请参阅FastAPI tutorial on Request Body。如果您试图避免使用Pydantic，或者出于某种原因不想为body创建类，那么还有一个Body类型参数：https://fastapi.tiangolo.com/tutorial/body-multiple-params/#singular-values-in-body。