标签: fastapi

我正在尝试http://httpbin.org/uuid使用以下代码片段一次向服务器发送 100 个请求

from fastapi import FastAPI
from time import sleep
from time import time
import requests
import asyncio

app = FastAPI()

URL= "http://httpbin.org/uuid"


# @app.get("/")
async def main():
    r = requests.get(URL)
    # print(r.text)
    
    return r.text

async def task():
    tasks = [main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main()]
    # print(tasks)
    # input("stop")
    result = await asyncio.gather(*tasks)
    print (result)

@app.get('/')
def f():
    start = time()
    asyncio.run(task())
    print("time: ",time()-start)

我将 FastAPI 与 Asyncio 结合使用，以实现大约 3 秒或更短的最短时间，但使用上述方法我得到的总时间为 66 秒，超过一分钟。我还想保留main用于附加操作的功能r.text。我知道要实现如此短的时间，需要并发性，但我不确定我在这里犯了什么错误。

假设我的main.py是这样的（这是一个简化的示例，在我的应用程序中我使用实际的数据库，并且有两个不同的数据库 URI 用于开发和测试）：

from fastapi import FastAPI
from pydantic import BaseSettings

app = FastAPI()

class Settings(BaseSettings):
    ENVIRONMENT: str

    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()

databases = {
    "dev": "Development",
    "test": "Testing"
}
database = databases[settings.ENVIRONMENT]

@app.get("/")
def read_root():
    return {"Environment": database}

而这.env是

ENVIRONMENT=dev

假设我想测试我的代码并且我想设置ENVIRONMENT=test为使用测试数据库。我应该怎么办？在FastAPI文档（https://fastapi.tiangolo.com/advanced/settings/#settings-and-testing）中有一个很好的例子，但它是关于依赖关系的，所以据我所知这是一个不同的情况。

我的想法如下（test.py）：

ENVIRONMENT=dev

但它不起作用。

此外，我收到此错误：

ScopeMismatch: You tried to access the 'function' scoped fixture 'monkeypatch' with a 'session' scoped request object, involved …

首先，我只想说这是我的第一个 Web 应用程序项目。在过去的几天里，我一直在努力寻找如何将前端和后端放在一起的答案。我有很多问题，但我想回答的主要问题是如何从后端端点返回我的前端“最终产品”。

我的理解是这样的（如有错误请指正）：

• 前端代码由客户端（浏览器）运行。
• 当客户端与网页交互时，前端根据需要对后端进行 API 调用以检索/修改数据。
• 后端和前端通常是单独开发的，并且可以托管在单独的服务器上。
• 但是，可以（并且可能更简单）将其托管在单个域/服务器上。我希望这样做，以避免 CORS 带来的一系列问题。

那么接下来的问题就来了：

当我想测试我的前端并看看它进展如何时，我只需运行npm run start. 然后，我访问给定的网址（通常http://localhost:8080/），并且可以访问我开发的前端。当我想部署它时，我运行npm run build，它给了我一个dist文件夹（捆绑在一起并缩小）。

如果我想在本地运行和测试我的后端，就像我正在使用的那样FastAPI，我只需运行uvicorn main:app --reload.

如何将两者放在一起？更具体地说，在我的后端代码中，如何返回前端工作的产品（即文件dist夹？）。我尝试过以下方法（简化）：

@app.get("/", response_class=HTMLResponse)
def root():
    return open("../frontend/dist/index.html", "r").read()

但是，当然，这只给了我静态 html，没有 React 组件。

我意识到这篇文章可能包含错误的假设和糟糕的做法（在这种情况下，我很抱歉！并且我将不胜感激任何更正/建议。）但是，如果可以回答以下问题，我将不胜感激。我提出的这些问题有望帮助我在计算机上本地测试整个 Web 应用程序。

1. 如何针对GET域根端点处的请求返回前端工作的结果？
2. 如果我的 Web 应用程序有页面 A、页面 B 和页面 C，每个页面都有 url www.example.com/A、www.example.com/B、 ，www.example.com/C我是否必须创建三个单独的 React 前端项目？即，相当于拥有三个dist文件夹？处理这个问题的标准方法是什么？

在处理 FastAPI 请求时，我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。

在 FastAPI 中执行此操作的正确方法是什么？我可以使用标准multiprocessing模块吗？到目前为止，我发现的所有教程/问题仅涵盖 I/O 绑定任务，如 Web 请求。

FastAPI 和 uvicorn 的新手，但我想知道为什么当我通过从命令行使用 uvicorn 启动它来运行我的“hello world”服务时，它工作正常，但是当从我的服务内部使用“uvicorn.run”方法时，服务启动，但是当我发送 GET 时，我总是得到一个404响应正文为{"detail": "Not Found"}?

这是我的代码：

import uvicorn
from fastapi import FastAPI

app = FastAPI()
uvicorn.run(app, host="127.0.0.1", port=5049)


@app.get("/")
async def root():
    return {"message": "Hello World"}

总是返回 404，如下所示：

# curl http://127.0.0.1:5049/
{"detail":"Not Found"}

我的服务的输出显示：

INFO:     Started server process [28612]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:5049 (Press CTRL+C to quit)
INFO:     127.0.0.1:55446 - "GET / HTTP/1.1" 404 Not Found

如果我注释掉“uvicorn.run”行，然后从命令行启动该服务（在 Windows 10 上运行）： …

对于 FastAPI Pydanctic 类，我有这些值

class ErrorReportRequest(BaseModel):
    sender: Optional[str] = Field(..., description="Who sends the error message.")
    error_message_displayed_to_client: str = Field(..., description="The error message displayed to the client.")

我使用该类作为输入模型

router = APIRouter()

@router.post(
    "/error_report",
    response_model=None,
    include_in_schema=True,

)
def error_report(err: ErrorReportRequest):
    pass

当我运行它时，sender是必填字段。如果它未包含在传入的 JSON 中，我会收到验证错误。

输入：

{
  "error_message_displayed_to_client": "string"
}

结果是：

{
  "detail": [
    {
      "loc": [
        "body",
        "sender"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

如果我像这样删除字段描述：

    class ErrorReportRequest(BaseModel):
        sender: Optional[str]
        error_message_displayed_to_client: str = Field(..., description="The error message displayed …

使用SQLModel如何让 alembic 识别以下模型？

from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

我一直在研究的一种方法是导入 Alembic 的 SQLalchemy 模型，但查看源代码我找不到如何做到这一点。

如何使 Alembic 与 SQLModel 模型一起使用？

一直在尝试使用 FASTAPI 中间件获取请求的正文，但似乎我只能获取 request.headers 而不能获取正文。我需要主体才能获得用于检查数据库中某些内容的密钥。考虑中间件的日志记录或身份验证使用。这必须在不自定义 APIRouter 的情况下完成，只需要在中间件级别完成，以免影响应用程序的其余部分。

@app.middleware("http")
    async def TestCustomMiddleware(request: Request, call_next):
    print("Middleware works!", request.headers)

    response = await call_next(request)
    resp_body = [section async for section in response.__dict__['body_iterator']]
    print("BODY:", resp_body)
    return response

我能够得到这个，但有一个错误会破坏 POST 请求：

INFO:     Started server process [37160]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Middleware works! Headers({'content-type': 'application/json', 'user-agent': 'PostmanRuntime/7.26.8', 'accept': '*/*', 'cache-control': 'no-cache', 'postman-token': 'ca6839ec-833d-45c0-9b52-8f904db13966', 'host': 'localhost:8000', 'accept-encoding': 'gzip, deflate, br', 'connection': 'keep-alive', 'content-length': '12'}) …

各位晚上好。我正在尝试使用 FastAPI 请求将新用户添加到我的数据库中。当我尝试通过 python 控制台应用程序执行此操作时，FastAPI 向我显示以下消息：

{
     'detail': [
         {
             'loc': ['body', 'nickname'],
             'msg': 'field required',
             'type': 'value_error.missing'
         },
         {
             'loc': ['body', 'password'],
             'msg': 'field required',
             'type': 'value_error.missing'
         },
         {
             'loc': ['body', 'email'],
             'msg': 'field required',
             'type': 'value_error.missing'
         }
    ]
}

但是，当我执行此请求时，/docs一切正常！

这是我的 pydantic 模型：

{
     'detail': [
         {
             'loc': ['body', 'nickname'],
             'msg': 'field required',
             'type': 'value_error.missing'
         },
         {
             'loc': ['body', 'password'],
             'msg': 'field required',
             'type': 'value_error.missing'
         },
         {
             'loc': ['body', 'email'],
             'msg': 'field required',
             'type': 'value_error.missing'
         } …

我有一个 FastAPI 应用程序，在多种不同的场合，需要调用外部 API。我使用 httpx.AsyncClient 进行这些调用。关键是我不完全理解应该如何使用它。

从httpx 的文档中我应该使用上下文管理器，

async def foo():
    """"
    I need to call foo quite often from different 
    parts of my application
    """
    async with httpx.AsyncClient() as aclient:
        # make some http requests, e.g.,
        await aclient.get("http://example.it")

但是，我明白，通过这种方式，每次我调用 时都会生成一个新客户端foo()，而这正是我们首先希望通过使用客户端来避免的情况。

我想另一种选择是在某处定义一些全局客户端，然后在需要时导入它，就像这样

aclient = httpx.AsyncClient()

async def bar():
    # make some http requests using the global aclient, e.g.,
    await aclient.get("http://example.it")

不过，第二个选项看起来有点可疑，因为没有人负责关闭会话等。

所以问题是：如何httpx.AsyncClient()在 FastAPI 应用程序中正确（重新）使用？