标签: fastapi

我正在 AWS ALB 后面部署一个快速 API 应用程序，侦听器规则路径模式/api/v1/指向快速 API。我的应用程序看起来像这样

from typing import Union
import os
import mysql.connector
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    print("Root path hit")
    return {"App": "Fargate"}


@app.get("/api/v1/")
def read_apiv1():
    print("Root path hit")
    return {"App": "Path Fargate API v1"}

我使用 docker 在 ECS 中部署了应用程序，我的 docker run 命令是

CMD ["uvicorn", "app.main:app", "--proxy-headers", "--host", "0.0.0.0", "--port", "80", "--root-path", "/api/v1"]

现在，当我点击后缀为 的 AWS ALB dns 时，我会看到抛出响应的/api/v1/端点。但是，根据快速 API 的文档，它应该使用./api/v1{"App": "Path Fargate API …

@router.post('/update_attributes/{username}')
def update_attributes(request: Request, username: str, email: str = Form(...)):
    pass

如何处理表单上输入的空字符串？email如果表单上的文本字段未填写，我会收到错误消息。

所以我有一个看起来像这样的请求

[\n    {\n        "Code": "EVR-T-0dsfdsdf532",\n        "Agent": "pacman",\n        "Kilometrage": "60000",\n        "Operation": "Vidange",\n        "Piece_Consomable": "filtre \xc3\xa0 air",\n        "Quantit\xc3\xa9": 1,\n        "UnitPrice": "200.00",\n        "Montant": 200,\n        "Mainoeuvre": 100\n    },\n    {\n        "Code": "EVR-T-ddsdf53dfds2",\n        "Agent": "pacman",\n        "Kilometrage": "60000",\n        "Operation": "Pneumatique",\n        "Piece_Consomable": "(Autre) Uiop",\n        "Quantit\xc3\xa9": 1,\n        "UnitPrice": "200.00"\n    }\n]\n

我的代码看起来像这样

@app.post("/utilities/Entretien/submit", status_code=status.HTTP_200_OK)\nasync def create_item(item: Request, db: Session = Depends(get_db)):\n    operations = await item.json()\n    for i in operations:\n        i : EntretienModel\n        new_operation = TableEntretien(**i)\n        db.add(new_operation)\n        db.commit()\n        db.refresh(new_operation)\n    return {"ping": "pong"}\n

我基本上是循环遍历数组，然后将每个对象插入数据库中，\n我正在寻找一种解决方案，可以使用如下所示的 pydantic 模型验证每个对象：

class …

我想将文件上传到 FastAPI 后端并将其转换为 Pandas DataFrame。但是，我似乎不明白如何使用 FastAPI 的UploadFile对象来做到这一点。更具体地说，我应该将什么传递给该pd.read_csv()函数？

这是我的 FastAPI 端点：

@app.post("/upload")
async def upload_file(file: UploadFile):
    df = pd.read_csv("")
    print(df)
    return {"filename": file.filename}

如何在 Fastapi 应用程序中对 API 端点请求进行速率限制？我需要为每个用户每秒限制 API 调用 5 个请求，并且超过该限制会阻止该特定用户 60 秒。

在 main.py

def get_application() -> FastAPI:
     application = FastAPI(title=PROJECT_NAME, debug=DEBUG, version=VERSION)
     application.add_event_handler(
        "startup", create_start_app_handler(application))
     application.add_event_handler(
        "shutdown", create_stop_app_handler(application))
     return application
app = get_application()

在事件.py

def create_start_app_handler(app: FastAPI) -> Callable:  
    async def start_app() -> None:           

        redis = await aioredis.create_redis_pool("redis://localhost:8080")
        FastAPILimiter.init(redis)
    return start_app

在端点

@router.post('/user',
             tags=["user"],
             name="user:user", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
***code****

从此文件 test.py 运行。

import uvicorn

from app.main import app

if __name__ == "__main__":
    uvicorn.run("test:app", host="0.0.0.0", port=8000, reload=True)

我如上编辑，但出现以下错误。

File "****ite-packages\starlette\routing.py", line 526, …