0

我试图找出验证使用 AWS API 网关创建并由 Python Lambda 函数支持的 API 的查询字符串参数的方法。API Gateway 可以验证是否存在所需的查询字符串参数。但是，我找不到其他验证的方法，例如确定某个参数的长度是否在某个限制范围内（例如 config_id 至少应为 7 个字符长）。使用 API Gateway 请求验证可以对请求正文进行此类验证。参考这个链接。但是，对于查询字符串参数，仅需要/不需要验证是可能的，因为它不使用任何 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 时，它不会返回任何错误并在 lambda 函数中继续进行。

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

4

1 回答 1

0

webargs库主要用于验证来自流行 Python 框架（如 Flask、Django、Bottle 等）的 HTTP 请求。您尝试使用的核心解析器不应直接使用，因为它没有实现 load_json、load_query 等方法（显示此处缺少的实现的源代码）。每个框架都有核心解析器的子类实现，但在 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,
        }
于 2020-12-07T19:25:54.340 回答