1

我有一个项目，其中包含一些使用 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().

现在，我需要调用insert_tab另一个函数，所以我需要以某种方式Request从Starlette实例化对象。我以前做过，但没有 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 主体注入到Request对象中，但我找不到这样做的方法。

有没有人这样做过或知道我该怎么做？

4

2 回答 2

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

于 2021-05-13T02:38:59.080 回答
0

我不是 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返回 a 的能函数Message，它是一种可变映射（例如 a dict）。原因receive是一个函数，因为它是“接收”通道的一部分，预计将“接收”请求正文作为消息流：

大多数有关传入请求的信息都存储在“范围”中，并在 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"（实际的正文/有效负载）和一个"more_body"键设置False为打破流。

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'sreceive=参数以调用您的函数：

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 dict

您实际上不需要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 教程。如果您想避免 Pydantic 或出于某种原因不想为 body 创建类，还有一个Body类型参数：https ://fastapi.tiangolo.com/tutorial/body-multiple-params/#singular-体内价值观。

于 2021-06-23T13:36:08.413 回答