如何通过httpx向FastAPI同时上传文件与JSON数据?
问题描述
我参考FastAPI官方文档编写了如下接口:
@app.post("/api/files") async def create_file( fileb: Annotated[UploadFile, File()], token: Annotated[str, Form()], ): return { "token": token, "fileb_content_type": fileb.content_type, }
使用httpx调用该接口,传递文件和字符串类型的表单参数时可以正常工作:
resp=httpx.post( "http://localhost:7071/api/files", files={'fileb': open('function.json','rb')}, data={'token':'fff'}, )
但实际业务中需要接收JSON格式数据,由于httpx的data参数不支持嵌套字典,我目前将JSON字典序列化后传递:
resp=httpx.post( "http://localhost:7071/api/files", files={'fileb': open('function.json','rb')}, data={'token':json.dumps({'y':'abc'})}, )
后端需要手动反序列化并转换为Pydantic的Params模型:
try: token = Params(**json.loads(token)) except Exception as err: return Response(f"Failed with {err}", 422)
这种实现方式感觉不够规范,想请教:
- 这是否是正确的实现方式?
- 能否在FastAPI接口参数中直接使用
Params模型? - 是否有办法在httpx请求中无需调用
json.dumps,可通过调整FastAPI或httpx请求实现?
解答
1. 这种实现是否正确?
可以运行,但不够规范。手动序列化/反序列化不仅增加冗余代码,还丢失了FastAPI自带的参数校验、自动文档生成等核心特性,同时错误处理逻辑也不够严谨(比如JSON格式错误场景,FastAPI原本可以自动返回标准化的422错误响应)。
2. 能否直接在FastAPI接口参数中使用Params模型?
完全可以。可以通过自定义依赖项或直接利用Pydantic的模型解析能力实现,以下是两种推荐方式:
方式一:直接在接口中解析JSON字符串为模型
from fastapi import Form, UploadFile, File, Response, HTTPException from pydantic import BaseModel, ValidationError # 定义Params模型 class Params(BaseModel): y: str @app.post("/api/files") async def create_file( fileb: Annotated[UploadFile, File()], token: Annotated[str, Form()], ): try: # 直接用Pydantic的模型方法解析JSON字符串 token_data = Params.model_validate_json(token) except ValidationError as err: # 返回标准化的422错误 raise HTTPException(status_code=422, detail=err.errors()) return { "token": token_data.dict(), "fileb_content_type": fileb.content_type, }
方式二:用依赖项封装解析逻辑(更简洁)
from fastapi import Depends # 自定义依赖项,负责解析表单中的JSON参数为Params模型 def parse_json_token(token: str = Form(...)): try: return Params.model_validate_json(token) except ValidationError as err: raise HTTPException(status_code=422, detail=err.errors()) @app.post("/api/files") async def create_file( fileb: Annotated[UploadFile, File()], # 直接将依赖项解析后的模型作为参数 token: Annotated[Params, Depends(parse_json_token)], ): return { "token": token.dict(), "fileb_content_type": fileb.content_type, }
3. 能否无需json.dumps,通过调整FastAPI或httpx实现?
可以,推荐两种方案:
方案一:封装httpx请求的表单数据处理
自定义工具函数自动处理嵌套字典的序列化,避免手动调用json.dumps:
import json import httpx def process_form_data(data: dict): processed = {} for key, value in data.items(): if isinstance(value, dict): processed[key] = json.dumps(value) else: processed[key] = value return processed # 调用时直接传递嵌套字典 resp = httpx.post( "http://localhost:7071/api/files", files={'fileb': open('function.json','rb')}, data=process_form_data({'token': {'y':'abc'}}), )
方案二:将JSON参数作为带媒体类型的表单字段传递
利用FastAPI支持解析特定媒体类型的表单字段特性,httpx将JSON参数作为application/json类型的表单项发送,后端直接绑定为Params模型:
后端接口调整
@app.post("/api/files") async def create_file( fileb: Annotated[UploadFile, File()], # 指定字段媒体类型为application/json,FastAPI自动解析为Params模型 token: Annotated[Params, File(media_type="application/json")], ): return { "token": token.dict(), "fileb_content_type": fileb.content_type, }
httpx请求代码
import json import httpx resp = httpx.post( "http://localhost:7071/api/files", files={ 'fileb': open('function.json','rb'), # 将JSON数据包装为带媒体类型的表单项,filename设为None 'token': (None, json.dumps({'y':'abc'}), 'application/json') } )
内容的提问来源于stack exchange,提问作者Dean MacGregor
相关产品推荐
相关产品推荐

