标签: fastapi

我有一个用Uvicorn + FastAPI编写的 REST-API 应用程序

我想使用 PyTest 进行测试。

我想在开始测试时在夹具中启动服务器，因此当测试完成时，夹具将终止应用程序。

FastAPI 测试展示了如何测试 API 应用程序，

from fastapi import FastAPI
from starlette.testclient import TestClient

app = FastAPI()


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


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

这不会以通常的方式使服务器联机。似乎由 client.get 命令触发的特定功能是唯一运行的东西。

我找到了这些额外的资源，但我无法让它们为我工作：

https://medium.com/@hmajid2301/pytest-with-background-thread-fixtures-f0dc34ee3c46

如何将服务器作为 py.test 的夹具运行

您将如何从 PyTest 运行 Uvicorn+FastAPI 应用程序，以便它随着测试而上升和下降？

我在 FastAPI 上有一个 API，当他请求我的页面时，我需要获取客户端的真实 IP 地址。

我很乐意使用starlette Request。但它返回我的服务器 IP，而不是客户端远程 IP。

我的代码：

@app.post('/my-endpoint')
async def my_endpoint(stats: Stats, request: Request):
    ip = request.client.host
    print(ip)
    return {'status': 1, 'message': 'ok'}

我在做什么错？如何获得真实IP（如在Flask request.remote_addr 中）？

我是非常新的 python 和 fastapi。我正在尝试在全局级别捕获未处理的异常。所以在main.py文件中的某个地方我写在下面：

@app.exception_handler(Exception)
async def exception_callback(request: Request, exc: Exception):
  logger.error(exc.detail)

但上述方法从未执行过。但是如果我编写一个自定义异常并尝试捕获它（如下所示），它运行良好。

class MyException(Exception):
  #some code

@app.exception_handler(MyException)
async def exception_callback(request: Request, exc: MyException):
  logger.error(exc.detail)

我已经完成了Catch 异常类型的 Exception 和 process body request #575。但是这个错误谈论访问请求正文。看到这个bug，感觉应该可以抓到Exception。FastApi 版本fastapi>=0.52.0。

提前致谢 ：）

FastAPI 使用 Gunicorn 启动 uvicorn 工作程序，如https://www.uvicorn.org/settings/中所述

但是，gunicorn 不允许使用自定义设置启动 uvicorn，如https://github.com/encode/uvicorn/issues/343中所述

该问题建议覆盖源文件中的 config_kwargs，例如https://github.com/encode/uvicorn/blob/master/uvicorn/workers.py

limit_concurrency我们尝试过，但 uvicorn 不遵守源中多个 uvicorn 文件中的设置：

https://github.com/encode/uvicorn/blob/master/uvicorn/workers.py

# fail

        config_kwargs = {
            "app": None,
            "log_config": None,
            "timeout_keep_alive": self.cfg.keepalive,
            "timeout_notify": self.timeout,
            "callback_notify": self.callback_notify,
            "limit_max_requests": self.max_requests, "limit_concurrency": 10000,
            "forwarded_allow_ips": self.cfg.forwarded_allow_ips,
        }

https://github.com/encode/uvicorn/blob/master/uvicorn/main.py

# fail

    kwargs = {
        "app": app,
        "host": host,
        "port": port,
        "uds": uds,
        "fd": fd,
        "loop": loop,
        "http": http,
        "ws": ws,
        "lifespan": lifespan,
        "env_file": env_file,
        "log_config": LOGGING_CONFIG if log_config is None else log_config,
        "log_level": log_level,
        "access_log": …

我正在尝试使用 FastAPI 为我们的组织创建 API。它有一个用于所有身份验证的 KeyCloak 服务器，以及被认为是最佳实践的 OpenID Connect 和 JWT。

在最简单的情况下，其他人负责获取有效的 JWT 令牌，以便 FastAPI 可以简单地解码和读取用户和权限。

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):

    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        jwt_token = jwt.decode(token, key=env.keycloak_server_public_key, audience='myorg')
        return jwt_token['preferred_username']
    except jwt.exceptions.ExpiredSignatureError:
        raise credentials_exception

生活是简单的！

但是，我确实想尝试让用户使用 Swagger 页面探索 API。我创建了这个函数，让用户使用 UI 登录：

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    login_request = requests.post(
        "https://mygreatorg.com/auth/realms/master/protocol/openid-connect/token",
        data={
            "grant_type": "password",
            "username": form_data.username,
            "password": form_data.password,
            "client_id": "fastapi-application",
        },
    )
    raw_response = json.loads(login_request.content.decode('utf-8'))
    raw_response['acquire_time'] …

我已经部署了一个 fastapi 端点，

from fastapi import FastAPI, UploadFile
from typing import List

app = FastAPI()

@app.post('/work/test')
async def testing(files: List(UploadFile)):
    for i in files:
        .......
        # do a lot of operations on each file

        # after than I am just writing that processed data into mysql database
        # cur.execute(...)
        # cur.commit()
        .......
    
    # just returning "OK" to confirm data is written into mysql
    return {"response" : "OK"}

我可以从 API 端点请求输出，它对我来说工作得很好。

现在，对我来说最大的挑战是知道每次迭代需要多少时间。因为在 UI 部分（那些访问我的 API 端点的人）我想帮助他们为正在处理的每个迭代/文件显示一个进度条（TIME TAKEN）。

我有什么可能的方法来实现它吗？如果是这样，请帮助我了解如何进一步处理？

谢谢你。

所以我正在学习 FastAPI，并试图弄清楚如何正确返回状态代码。我创建了一个用于上传文件的端点，并且我想在文件格式不受支持的情况下做出特殊响应。似乎我按照官方文档做了一切，但我总是收到422 Unprocessable Entity错误。

这是我的代码：

from fastapi import FastAPI, File, UploadFile, status
from fastapi.openapi.models import Response
    
app = FastAPI()
    
@app.post('/upload_file/', status_code=status.HTTP_200_OK)
async def upload_file(response: Response, file: UploadFile = File(...)):
    """End point for uploading a file"""
    if file.content_type != "application/pdf":
        response.status_code = status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
        return {f'File {file.filename} has unsupported extension type'}

    return {'filename': file.content_type}

先感谢您！

我刚刚开始使用 FastAPI，但我不知道如何为 Pydantic 模型编写单元测试（使用 pytest）。

这是 Pydantic 模型的示例：

class PhoneNumber(BaseModel):
    id: int
    country: str
    country_code: str
    number: str
    extension: str

我想通过创建示例PhoneNumber实例来测试此模型，并确保该PhoneNumber实例与字段类型相符。例如：

PhoneNumber(1, "country", "code", "number", "extension")

然后，我想断言 PhoneNumber.country 等于“country”。

我希望我的 FastAPI 应用程序在出现以下情况时重新加载.csv我希望我的 FastAPI 应用程序在同一目录中的文件发生更改

我尝试了以下命令，但它不起作用。

uvicorn main:app --reload --reload-include *.csv

有谁知道这个问题的解决方案？

我正在尝试使用文档页面传递授权标头，类似于此页面：

由于文档是在 Fast API 中自动生成的，因此我很难弄清楚这一点。我关注了此页面https://fastapi.tiangolo.com/tutorial/security/但找不到有关传递不记名令牌的任何信息。请注意，我并不是在寻找验证令牌，我只是在寻找一种通过文档页面传递不记名令牌的方法。

任何人都可以参考一些相关文档或寻求帮助。