使用webargs验证AWS lambda中的查询字符串参数和请求主体

Question

我正试图找出验证使用API网关创建并由Python函数支持的API的查询字符串参数的方法。 API网关可以验证所需查询字符串参数的存在。但是，我无法找到其他验证的方法，例如确定某个参数的长度是否在某个限制范围内(例如，config_id应该至少有7个字符长)。这样的验证对于使用API网关请求验证的请求主体是可能的。参考这链接。但是，对于查询字符串参数，只需要/不需要验证是可能的，因为它不使用任何json模式进行验证。

因此，为了克服这个问题，我决定在Python中使用webargs模块来验证查询字符串参数。它通常用于对使用Python框架(如烧瓶或django )创建的API进行请求验证。我正在使用核心解析器(参见韦格斯医生)，如下所示：

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时，它不会返回任何错误，并在lambda函数中继续进行。

这会有什么问题呢？解析器似乎可以工作，但是验证功能不起作用。任何帮助都是非常感谢的，因为我对此并不熟悉。另外，在lambda函数中是否有更好的方法来进行验证，特别是对于queryStringParameters？它可以由代码来处理，但是我们可以有很多参数和很多API，这使得为所有这些验证编写代码成为一项繁琐的任务。这个webargs模块非常有用。

Answer 1

韦布格库主要用于验证通过诸如load_json、Django、does等流行的load_query框架发出的HTTP请求。您试图使用的核心解析器不应该直接使用，因为它没有实现load_json、load_query等方法(源代码显示缺少的实现这里)。每个框架都有核心解析器的子类实现，但是在API上使用它们没有意义。

所以最好使用一个更简单的json验证库，比如jsonschema。我已经修改了您的代码以使用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,
        }