使用 webargs 在 AWS lambda 中验证查询字符串参数和请求正文

Validating query string parameters and request body in AWS lambda using webargs

我正在尝试为使用 AWS API 网关创建并由 Python Lambda 函数支持的 API 找出验证查询字符串参数的方法。 API 网关可以验证所需的查询字符串参数是否存在。但是,我找不到进行额外验证的方法,例如确定某个参数的长度是否在某个限制内(例如 config_id 应该至少有 7 个字符长)。使用 API 网关请求验证可以对请求正文进行此类验证。参考 this link。但是,对于查询字符串参数,只有 required/not 需要验证是可能的,因为它不使用任何 json 模式进行验证。

因此,为了克服这个问题,我决定尝试 Python 中的 webargs 模块来验证查询字符串参数。它通常用于对使用 Python 框架(如 flask 或 django)创建的 API 的请求验证。我正在使用核心解析器(参考webargs doc)如下:

from webargs import fields, validate, core, ValidationError
parser = core.Parser()

params = {"config_id": fields.Str(required=True, validate=lambda p: len(p) >= 7)}

def main(event, context: Dict):
    try:
        # print(event["queryStringParameters"])
        input_params = event.get("queryStringParameters")
        print("queryStringParameters: ", str(input_params))

        if input_params is None:
            input_params = {}
        parsed_params = parser.parse(params, input_params)
        print("parsedParams: ", str(parsed_params))
    except ValidationError as e:
        return {
            "statusCode": 400,
            "headers": {
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Credentials": True,
                "x-amzn-ErrorType": "ValidationError",
            },
            "body": str(e),
        }

这是在 lambda 函数中完成验证的方式。但是,只有必需的验证才能正常工作。当我传递长度为 5 的 config_id 时,它不会 return 任何错误并在 lambda 函数中进一步处理。

这可能出了什么问题? 解析器似乎可以工作,但是验证函数却没有。 感谢任何帮助,因为我是新手。此外,是否有更好的方法在 lambda 函数中进行验证,尤其是对于 queryStringParameters?它可以由代码来处理,但是我们可以有很多参数和很多API,这使得为所有这些验证编写代码成为一项繁琐的任务。 webargs 模块派上用场。

webargs Library is mostly used for validating HTTP Requests coming via popular Python frameworks like Flask, Django, Bottle etc. The core Parser that you are trying to use should not be used directly as it does not have the methods like load_json, load_query etc implemented (Source code showing the missing implementation here)。每个框架都有核心解析器的子 class 实现,但在 API GW 上使用它们没有意义。

所以最好使用像 jsonschema 这样更简单的 json 验证库。我修改了您的代码以使用 jsonschema 而不是 webargs,如下所示 -

from jsonschema import validate, ValidationError

schema = {
     "type" : "object",
     "properties" : {
         "queryStringParameters" : {
                "type" : "object",
                "properties": {
                        "config_id": {
                                "type": "string",
                                "minLength": 7,
                        }
                }
        },

     },
 }


def main(event, context):
    try:
        validate(instance=event, schema=schema)
    except ValidationError as e:
        return {
            "statusCode": 400,
            "headers": {
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Credentials": True,
                "x-amzn-ErrorType": "ValidationError",
            },
            "body": e.message,
        }