fastapi 打印 post 的 request 的 form?
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/
回复
1个回答
test
2024-06-28
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
)
回复
适合作为回答的
- 经过验证的有效解决办法
- 自己的经验指引,对解决问题有帮助
- 遵循 Markdown 语法排版,代码语义正确
不该作为回答的
- 询问内容细节或回复楼层
- 与题目无关的内容
- “赞”“顶”“同问”“看手册”“解决了没”等毫无意义的内容