测试触发CloudFront缓存失效的S3触发Lambda时为何出现KeyError？

场景描述 #

用户编写了一个Python 3.7 Lambda函数用于触发CloudFront缓存失效，配置了对应的IAM策略，但使用自定义测试JSON调用时，函数抛出KeyError: 'Records'错误。

Lambda函数代码 #

from __future__ import print_function

import boto3
import time

def lambda_handler(event, context):
    print(event)
    for items in event["Records"]:
        path = "/" + items["s3"]["object"]["key"]
    print(path)
    client = boto3.client('cloudfront')
    invalidation = client.create_invalidation(DistributionId='distributionid__ID',
        InvalidationBatch={
            'Paths': {
                'Quantity': 1,
                'Items': [path]
        },
        'CallerReference': str(time.time())
    })

关联的IAM策略 #

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "cloudfront:CreateInvalidation"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}

测试调用JSON #

{
  "key1": "*"
}

触发的错误信息 #

{'key1': '*'}
[ERROR] KeyError: 'Records'
Traceback (most recent call last):
Version: $LATEST
    File "/var/task/lambda_function.py", line 8, in lambda_handler
        for items in event["Records"]:

问题 #

请问为何会出现这个KeyError？

原因分析 #

你的Lambda函数代码是为S3事件触发场景编写的：当S3存储桶有对象更新时，S3会向Lambda发送包含Records数组的事件数据，代码依赖这个Records字段来获取S3对象路径。但你当前使用的测试JSON里只有key1键，完全没有Records字段，代码执行到event["Records"]时找不到对应键，自然抛出KeyError。

修复方案 #

根据你的实际需求选择对应的修复方式：

需求1：仅用S3事件触发Lambda #

• 放弃当前的自定义测试JSON，改用AWS Lambda控制台提供的S3 Put事件测试模板，该模板的结构包含Records字段，完全匹配代码的预期格式。示例模板结构如下：

{
  "Records": [
    {
      "s3": {
        "object": {
          "key": "example-object-key"
        }
      }
    }
  ]
}

需求2：支持自定义输入（手动指定缓存失效路径） #

• 修改Lambda函数代码，同时兼容S3事件和自定义输入格式，示例代码如下：

from __future__ import print_function

import boto3
import time

def lambda_handler(event, context):
    print(event)
    path = ""
    
    # 处理自定义输入逻辑
    if "key1" in event:
        path = event["key1"]
        # 确保路径格式正确：非*的路径开头添加/
        if path != "*" and not path.startswith("/"):
            path = "/" + path
    # 处理S3事件逻辑
    elif "Records" in event and len(event["Records"]) > 0:
        items = event["Records"][0]
        path = "/" + items["s3"]["object"]["key"]
    else:
        raise ValueError("不支持的事件格式，请检查输入")
    
    print(path)
    client = boto3.client('cloudfront')
    invalidation = client.create_invalidation(
        DistributionId='distributionid__ID',
        InvalidationBatch={
            'Paths': {
                'Quantity': 1,
                'Items': [path]
            },
            'CallerReference': str(time.time())
        }
    )

修改后，既可以用{"key1": "*"}测试全量缓存失效，也能正常响应S3事件触发的单路径失效。

内容的提问来源于stack exchange，提问作者Santosh