如何在APIFlask中处理包含文件的多部分请求并分别验证JSON与文件部分
如何在APIFlask中处理包含文件的多部分请求并分别验证JSON与文件部分
嘿,这个问题我刚好踩过坑,现在来给你捋清楚解决方案!你遇到的核心限制确实是APIFlask不允许给一个视图函数同时声明多个请求体位置,但我们可以通过form_and_files这个位置来绕开限制,同时实现JSON和文件部分的分开验证,下面给你两种实用的方案:
方案一:手动解析+分步验证
这种方式直观易懂,先通过form_and_files接收完整的多部分请求,再单独处理JSON的解析和验证:
首先定义你需要的JSON数据Schema:
from apiflask import Schema from marshmallow import fields, validate class DataSchema(Schema): # 这里根据你的实际业务需求定义JSON字段规则 name = fields.Str(required=True, validate=validate.Length(min=1)) age = fields.Int(required=True, validate=validate.Range(min=1))
再定义包含JSON字段和文件字段的整体请求Schema:
class MultipartRequestSchema(Schema): json_payload = fields.Str(required=True, description="JSON格式的请求数据") upload_file = File(required=True, description="需要上传的文件")
接下来编写视图函数:
from apiflask import APIFlask, input, File import json app = APIFlask(__name__) v1 = app.create_api_blueprint('v1', url_prefix='/v1') @v1.post("/data") @v1.input(MultipartRequestSchema, location="form_and_files") def post_data(request_data): # 第一步:解析并验证JSON部分 try: # 把form中的JSON字符串转成字典格式 json_data = json.loads(request_data['json_payload']) # 用DataSchema验证JSON数据的合法性 validated_json = DataSchema().load(json_data) except json.JSONDecodeError: return {"message": "传入的JSON格式无效"}, 400 except Exception as e: return {"message": f"JSON数据验证失败:{str(e)}"}, 400 # 第二步:处理文件部分 uploaded_file = request_data['upload_file'] # 这里可以添加文件专属验证,比如检查文件类型、大小 if not uploaded_file.filename.endswith('.txt'): return {"message": "仅支持上传TXT格式文件"}, 400 # 后续业务逻辑,比如保存文件、处理数据 # uploaded_file.save(f"./uploads/{uploaded_file.filename}") return { "validated_json": validated_json, "file_info": { "filename": uploaded_file.filename, "content_type": uploaded_file.content_type } }, 200
方案二:自定义字段封装验证逻辑
如果觉得手动解析太繁琐,我们可以自定义一个Marshmallow字段,把JSON的解析和验证逻辑封装进去,让代码更简洁优雅:
先定义自定义字段:
from marshmallow.fields import Field import json class JSONSchemaField(Field): def __init__(self, schema_cls, **kwargs): self.schema = schema_cls() super().__init__(**kwargs) def _deserialize(self, value, attr, data, **kwargs): # 第一步:解析JSON字符串 try: json_dict = json.loads(value) except json.JSONDecodeError: raise self.make_error("invalid_json_format", message="JSON格式错误") # 第二步:用指定Schema验证数据 try: return self.schema.load(json_dict) except Exception as e: raise self.make_error("json_validation_failed", message=str(e))
然后修改MultipartRequestSchema,用自定义字段替代普通字符串字段:
class MultipartRequestSchema(Schema): json_payload = JSONSchemaField(DataSchema, required=True, description="JSON格式的请求数据") upload_file = File(required=True, description="需要上传的文件")
最后视图函数就能简化成这样:
@v1.post("/data") @v1.input(MultipartRequestSchema, location="form_and_files") def post_data(request_data): # 这里的json_payload已经是验证后的合法字典了 validated_json = request_data['json_payload'] uploaded_file = request_data['upload_file'] # 后续业务逻辑... return { "validated_json": validated_json, "filename": uploaded_file.filename }, 200
小提示
- 前端发送请求时,要把JSON数据以字符串形式放在form字段中(比如字段名设为
json_payload),同时携带文件字段,这样APIFlask才能通过form_and_files正确接收。 - 如果需要限制文件大小,可以给
File()字段添加验证逻辑,比如:File(required=True, validate=lambda f: len(f.read()) <= 1024*1024)(注意验证后要把文件指针移回开头:f.seek(0))。
备注:内容来源于stack exchange,提问作者George Boukeas




