标签: fastapi

from fastapi import FastAPI
from fastapi.params import Body

app = FastAPI()

@app.post("/createposts")
def create_posts(payload: dict = Body(...)):
    print(payload)
    return {"new_post" : f"title {payload["title"]} content: {payload["content"]}"}

我正在尝试使用 Fastapi 创建 API，但每次运行代码时都会收到与 return 语句相关的错误： SyntaxError: f-string: unmatched '['

谢谢你！

我有一个在我的所有 API 中作为依赖项调用的函数。

from fastapi import HTTPException, Request, status


async def get_user_id_or_401(request: Request) -> str:
    user_id: str = request.headers.get("x-cognito-user-id")

    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
        )
    return user_id

如何对该功能进行单元测试？我最初的举动是修补request.headers.get，但我不知道该怎么做。

所以，我的问题是，如何修补request.headers.get. 但是，如果我有更好的方法可以对此进行测试，请告诉我。

我想将图像添加到 FastAPI 自动文档（由 Swagger UI 提供），但我不知道如何执行此操作。这是代码：

@api.get(path='/carbon-credit/',
    responses={
        200: {'description': 'Ok',
            "content": {
            "image/jpeg": {
                "example": 'https://picsum.photos/seed/picsum/200/300'
                    }
                }},
        404: {"description": "not found"},
        422: {'description': 'not found 2'},
    },
    name='API for Carbon Credit',
    description="get carbon credit",
    tags=['Images'],
    response_class=Response)

正如您从代码中看到的，我尝试使用 URL 来执行此操作，而我在 ReDoc 和 Swagger UI 中得到的只是文本形式的 URL，而不是实际图像。另外，我想使用存储在本地驱动器中的图像。

Swagger UI 和 ReDoc 的屏幕截图：

我怎样才能做到这一点？

提前致谢。

我正在尝试测试将文件上传到 FastApi，但我不断收到 422 验证错误。文件上传在OpenApi接口下可以正常上传，但在下面的测试中不行。

路由器：

@router.post("/files")
def file_contents(files: List[UploadFile]):
    return someprocessing(files)

测试（使用TestClientFastApi）：

response = client.post(
    url="/files",
    files={"files": ("file.xlsx", open("./test_files/file.xlsx", "rb"))},
    headers={**auth_headers, **{"Content-Type": "multipart/form-data"}},
)

错误：

{"detail":[{"loc":["body"],"msg":"value is not a valid dict","type":"type_error.dict"}]}

更新：很好，我正在将文件发送到不同的网址...

我知道这不是 FastAPI 问题，但如何使用 FastAPI 避免这种情况？

例如：

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def root(q: str):
    return {"message": f"{q}"}

发出以下请求：

http://127.0.0.1:8000/?q=1+1

返回：

{"message":"1 1"}

我有一个非常简单的 FastAPI-Traefik-Docker 设置，您可以在这里找到： https: //github.com/mshajarrazip/fastapi-traefik-docker-cors

只需docker-compose up -d运行 FastAPI 和 traefik 服务即可。

此设置中未配置 TLS/SSL。

该应用程序有两个端点 GET“/hello”和 POST“/add”，它们除了返回 json 响应之外什么也不做：

在app/main.py

import logging\n\nfrom app import routes\nfrom fastapi import FastAPI\nfrom fastapi.middleware.cors import CORSMiddleware\n\ndef create_application() -> FastAPI:\n    app = FastAPI(title="\xe2\x9c\xa8 ML API \xe2\x9c\xa8")\n\n    app.include_router(routes.hello.router)\n\n    # add CORS\n    # origins = ["*"]\n    app.add_middleware(\n        CORSMiddleware,\n        allow_origins=["*"], # Allows all origins\n        allow_credentials=True,\n        allow_methods=["*"], # Allows all methods\n        allow_headers=["*"], # Allows all headers\n        )\n\n    return app\n\n\napp = create_application()\n

我正在 windows/ubuntu 中的 docker 容器内开发 fastapi （代码如下）。当我通过在终端中运行python -m uvicorn app:app --reload测试容器外部的应用程序时，然后导航到127.0.0.1:8000/home一切正常：

{
  Data: "Test"
}

但是，当我docker-compose up时，我既无法在容器中运行python -m uvicorn app:app --reload（由于端口已被使用），也无法在浏览器中看到任何返回的内容。我尝试过 127.0.0.1:8000/home、host.docker.internal:8000/home 和 localhost:8000/home 并且我总是收到：

{
   detail: "Not Found"
}

我缺少哪一步？

Dockerfile：

FROM python:3.8-slim

EXPOSE 8000

ENV PYTHONDONTWRITEBYTECODE=1

ENV PYTHONUNBUFFERED=1

COPY requirements.txt .
RUN python -m pip install -r requirements.txt

WORKDIR /app
COPY . /app

RUN adduser -u nnnn --disabled-password --gecos "" appuser && chown -R appuser /app
USER appuser

CMD ["gunicorn", …

我正在开发一项 python 服务，其中我必须创建一个 FastAPI 休息端点，并且应该在后台运行 2 个调度程序任务。

我们如何在 FastAPI 中实现调度程序？如果不可行，实现这种要求的最佳方法是什么，创建两个服务是正确的方法？

谢谢和问候，阿鲁

我使用 FastAPI 构建了一个 API，用于从 Google Cloud Storage 中的存储桶中检索数据。我使用 swagger 和 Postman 在本地测试了 API（进行双重检查），它工作得很好，但是当我使用 docker 对其进行容器化时，我收到一个无效的存储桶名称：“None”，400 错误。

这是代码...

import os
import gcsfs
from typing import Any, List

import pandas as pd
from dotenv import load_dotenv
from fastapi import APIRouter

from app import schemas 
from app.config import settings

load_dotenv()

GOOGLE_SERVICE_ACCOUNT = os.getenv("GOOGLE_SERVICE_ACCOUNT")
GCP_FILE_PATH = os.getenv("GCP_FILE_PATH")
fs = gcsfs.GCSFileSystem()

api_router = APIRouter()

@api_router.get("/health", response_model=schemas.Health, status_code=200)
def health() -> dict:
    """
    Root Get
    """
    health = schemas.Health(
        name=settings.PROJECT_NAME, api_version="1.0.0"
    )

    return health.dict()

@api_router.get("/consumer_type_values", response_model=schemas.UniqueConsumerType, …

我想将我的 FastAPI 代码分发到多个文件中，以便团队清楚地了解，因此我尝试使用 APIRoute 代替装饰器，但这样我就无法在其他文件中使用 FastAPI 装饰器。可以说我有main.py这样的

... import statements

rts = [
    APIRoute("/{value}", methods=["get"], endpoint=read_item)
]

app = FastAPI(routes=rts)

uvicorn.run("main:app", host="0.0.0.0", port=80)

read_item现在我已经在另一个名为read_item.py我在主文件中导入的文件中定义了方法。

def read_item(value: int | float = None):
    results = {"value": value}
    return results

如果我开始在主文件中使用@app装饰器，它会变得太大。现在的问题是我无法访问文件@app.on_event中的应用程序装饰器read_items.py，因为应用程序是在main文件中声明的，并且方法需要在read_item文件中声明。FastAPI 有什么办法解决这个问题吗？