You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何在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

火山引擎 最新活动