python-aws-lambda-handlers

一个有偏见的Python包，它简化了指定AWS Lambda处理器的过程，包括输入验证、错误处理和响应格式化。

它是100%开源的，并使用APACHE2许可。

快速入门

安装

要安装lambda-handlers的最新版本，只需运行

pip install lambda-handlers

如果您要使用验证，我们有一些示例，它们与Marshmallow或jsonschema一起工作。

但是，您可以将LambdaHandler调整为使用您喜欢的验证模块。如果您需要任何帮助，请与我们分享或创建一个问题。

默认情况下，http_handler装饰器确保将请求体解析为JSON，并将响应格式化为JSON，包括：- 适当的statusCode，- CORS头，以及- 处理器返回值在主体中。

基本的ApiGateway代理处理器

from lambda_handlers.handlers import http_handler

@http_handler()
def handler(event, context):
    return event['body']

示例

HTTP处理器

跳过默认的CORS头并配置它。

from lambda_handlers.handlers import http_handler
from lambda_handlers.response import cors

@http_handler(cors=cors(origin='localhost', credentials=False))
def handler(event, context):
    return {
        'message': 'Hello World!'
    }

aws lambda invoke --function-name example response.json
cat response.json

{
    "headers":{
        "Access-Control-Allow-Origin": "localhost",
        "Content-Type": "application/json"
    },
    "statusCode": 200,
    "body": "{\"message\": \"Hello World!\"}"
}

验证

使用jsonschema验证用户模型作为输入。

from typing import Any, Dict, List, Tuple, Union

import jsonschema

from lambda_handlers.handlers import http_handler
from lambda_handlers.errors import EventValidationError

class SchemaValidator:
    """A payload validator that uses jsonschema schemas."""

    @classmethod
    def validate(cls, instance, schema: Dict[str, Any]):
        """
        Raise EventValidationError (if any error) from validating 
        `instance` against `schema`.
        """
        validator = jsonschema.Draft7Validator(schema)
        errors = list(validator.iter_errors(instance))
        if errors:
            field_errors = sorted(validator.iter_errors(instance), key=lambda error: error.path)
            raise EventValidationError(field_errors)

    @staticmethod
    def format_errors(errors: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Re-format the errors from JSONSchema."""
        path_errors: Dict[str, List[str]] = defaultdict(list)
        for error in errors:
            path_errors[error.path.pop()].append(error.message)
        return [{path: messages} for path, messages in path_errors.items()]

user_schema: Dict[str, Any] = {
    'type': 'object',
    'properties': {
        'user_id': {'type': 'number'},
    },
}

@http_handler()
def handler(event, context):
    user = event['body']
    SchemaValidator.validate(user, user_schema)
    return user

aws lambda invoke --function-name example --payload '{"body": {"user_id": 42}}' response.json
cat response.json

{
    "headers":{
        "Access-Control-Allow-Credentials": true,
        "Access-Control-Allow-Origin": "*",
        "Content-Type": "application/json"
    },
    "statusCode": 200,
    "body": "{\"user_id\": 42}"
}

使用Marshmallow验证用户模型作为输入体和响应体。

from typing import Any, Dict, List, Tuple, Union

from marshmallow import Schema, fields, ValidationError

from lambda_handlers.handlers import http_handler
from lambda_handlers.errors import EventValidationError

class SchemaValidator:
    """A data validator that uses Marshmallow schemas."""

    @classmethod
    def validate(cls, instance: Any, schema: Schema) -> Any:
        """Return the data or raise EventValidationError if any error from validating `instance` against `schema`."""
        try:
            return schema.load(instance)
        except ValidationError as error:
            raise EventValidationError(error.messages)

class UserSchema(Schema):
    user_id = fields.Integer(required=True)

@http_handler()
def handler(event, context):
    user = event['body']
    SchemaValidator.validate(user, UserSchema())
    return user

aws lambda invoke --function-name example --payload '{"body": {"user_id": 42}}' response.json
cat response.json

{
    "headers":{
        "Access-Control-Allow-Credentials": true,
        "Access-Control-Allow-Origin": "*",
        "Content-Type": "application/json"
    },
    "statusCode": 200,
    "body": "{\"user_id\": 42}"
}

aws lambda invoke --function-name example --payload '{"body": {"user_id": "peter"}}' response.json
cat response.json

{
    "headers":{
        "Access-Control-Allow-Credentials": true,
        "Access-Control-Allow-Origin": "*",
        "Content-Type": "application/json"
    },
    "statusCode": 400,
    "body": "{\"errors\": {\"user_id\": [\"Not a valid integer.\"]}"
}

头部

Cors

from lambda_handlers.handlers import http_handler
from lambda_handlers.response import cors

@http_handler(cors=cors(origin='example.com', credentials=False))
def handler(event, context):
    return {
        'message': 'Hello World!'
    }

aws lambda invoke --function-name example response.json
cat response.json

{
    "headers":{
        "Access-Control-Allow-Origin": "example.com",
        "Content-Type": "application/json"
    },
    "statusCode": 200,
    "body": "{\"message\": \"Hello World!\"}"
}

错误

LambdaHandlerError

BadRequestError

ForbiddenError

InternalServerError

NotFoundError

FormatError

ValidationError

分享爱心

喜欢这个项目？请在我们GitHub上给它一个★！

相关项目

查看这些相关项目。

• node-aws-lambda-handlers - 一个有偏见的TypeScript包，它简化了指定AWS Lambda处理器的过程，包括输入验证、错误处理和响应格式化。

帮助

有问题吗？

在GitHub上提交问题。

贡献

错误报告 & 功能请求

请使用问题跟踪器报告任何错误或提交功能请求。

开发

如果您有兴趣成为贡献者并想参与这个项目的开发，我们非常愿意听到您的意见！

一般来说，PR是受欢迎的。我们遵循典型的“分叉和拉取”Git工作流程。

1. 在GitHub上分叉存储库
2. 将项目克隆到您的计算机上
3. 将更改提交到您自己的分支
4. 将您的工作推回到您的分叉
5. 提交一个拉取请求，以便我们可以审查您的更改

注意：在提交拉取请求之前，请确保合并来自“上游”的最新更改！

许可证

有关详细信息，请参阅LICENSE。

Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  https://apache.ac.cn/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.