fastapi 打印 post 的 request 的 form？

Z时代
2024-03-04
分类：IT

fastapi 如何打印 post 的 request 的 form ？以字典的方式打印所有 form 表单参数

我先写了下面这玩意，请求直接报错

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.requests import Request
from fastapi import FastAPI, Form, Query, Depends
from fastapi import Depends
from pydantic import Field
from datetime import datetime
from typing import Any, Optional, List
import uvicorn
from loguru import logger
app = FastAPI()
# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=["*"],
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )
# Swagger.doc(app)
@app.post('/search')
def search_reverse(
    request: Request,
    max_chunk_size: int = Form(1200, description='分批查询 milvus，每个查询批次的向量个数'),
    search_top_k: int = Form(
        default=30, description='搜索 milvus 的时候，返回的最相似的 N 个结果'),
    distance_threshold: float = Form(
        0.32, description='距离上限（距离使用 L2 欧氏距离衡量，越接近 0 越相似），但 milvus 的搜索结果返回后，丢弃大于该值的向量搜索结果')
):
    logger.debug(request._form())
if __name__ == "__main__":
    uvicorn.run(
        app='api:app',
        host="127.0.0.1",
        port=9950,
        workers=1,
        reload=True
    )

请求报错如下：

  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/routing.py", line 237, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/routing.py", line 165, in run_endpoint_function
    return await run_in_threadpool(dependant.call, **values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/concurrency.py", line 41, in run_in_threadpool
    return await anyio.to_thread.run_sync(func, *args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/Desktop/code/me/fastapi_example/api.py", line 41, in search_reverse
    logger.debug(request._form())
                 ^^^^^^^^^^^^^^^TypeError: 'FormData' object is not callable

声明，我要同步，不要异步！所以不可以往视图函数前面加 async

然后我按照 chatGPT 的提示，改成下面那样，还是报错

from fastapi import FastAPI, Form, Request
from loguru import logger
import uvicorn
app = FastAPI()
@app.post('/search')
def search_reverse(
    request: Request,
    max_chunk_size: int = Form(1200, description='分批查询 milvus，每个查询批次的向量个数'),
    search_top_k: int = Form(
        default=30, description='搜索 milvus 的时候，返回的最相似的 N 个结果'),
    distance_threshold: float = Form(
        0.32, description='距离上限（距离使用 L2 欧氏距离衡量，越接近 0 越相似），但 milvus 的搜索结果返回后，丢弃大于该值的向量搜索结果')
):
    form_data = request.form()
    form_dict = dict(form_data)
    logger.debug(form_dict)
if __name__ == "__main__":
    uvicorn.run(
        app='api:app',
        host="127.0.0.1",
        port=9950,
        workers=1,
        reload=True
    )

报错

  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 21, in __call__
    raise e
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
    await self.app(scope, receive, send)
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
    await route.handle(scope, receive, send)
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/routing.py", line 237, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/routing.py", line 165, in run_endpoint_function
    return await run_in_threadpool(dependant.call, **values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/starlette/concurrency.py", line 41, in run_in_threadpool
    return await anyio.to_thread.run_sync(func, *args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/Desktop/code/me/fastapi_example/api.py", line 17, in search_reverse
    form_dict = dict(form_data)
                ^^^^^^^^^^^^^^^
TypeError: 'AwaitableOrContextManagerWrapper' object is not iterable

下面这样写更不行

@app.post('/search')
def search_reverse(
    form: Form,
    max_chunk_size: int = Form(1200, description='分批查询 milvus，每个查询批次的向量个数'),
    search_top_k: int = Form(
        default=30, description='搜索 milvus 的时候，返回的最相似的 N 个结果'),
    distance_threshold: float = Form(
        0.32, description='距离上限（距离使用 L2 欧氏距离衡量，越接近 0 越相似），但 milvus 的搜索结果返回后，丢弃大于该值的向量搜索结果')):

直接报错

  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/dependencies/utils.py", line 462, in analyze_param
    field = create_response_field(
            ^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ponponon/.local/share/virtualenvs/fastapi_example-YHDlb6MG/lib/python3.11/site-packages/fastapi/utils.py", line 87, in create_response_field
    raise fastapi.exceptions.FastAPIError(fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <function Form at 0x1063282c0> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/

回答：

from fastapi import FastAPI, Form
from loguru import logger
import uvicorn
app = FastAPI()
@app.post("/search")
def search_reverse(**kwargs: int):
    logger.debug(kwargs)
if __name__ == "__main__":
    uvicorn.run(
        app='api:app',
        host="127.0.0.1",
        port=9950,
        workers=1,
        reload=True
    )

或者：

from fastapi import FastAPI, Form, Depends, Request
from loguru import logger
import uvicorn
app = FastAPI()
async def get_form(request: Request):
    form_data = await request.form()
    return dict(form_data)
@app.post("/search")
def search_reverse(form_data: dict = Depends(get_form)):
    logger.debug(form_data)
if __name__ == "__main__":
    uvicorn.run(
        app='api:app',
        host="127.0.0.1",
        port=9950,
        workers=1,
        reload=True
    )

以上是 fastapi 打印 post 的 request 的 form？的全部内容，来源链接： utcz.com/p/938998.html

fastapi 打印 post 的 request 的 form？

回答：

其他人也看了：