CDK Python中如何捕获aws s3 cp命令触发的S3事件？

首先明确aws s3 cp触发的S3事件规则：

• 上传文件小于8MB时，默认触发OBJECT_CREATED_PUT事件
• 上传文件大于等于8MB时，aws cli默认开启分片上传，完成后触发OBJECT_CREATED_COMPLETE_MULTIPART_UPLOAD事件

你当前已经配置了这两类事件，理论上可以覆盖所有aws s3 cp的上传场景，无法捕获事件可按以下顺序排查：

1. 确认Lambda调用权限配置正确 #

S3触发Lambda需要Lambda侧配置基于资源的权限，允许对应S3桶调用Lambda函数。很多场景下CDK不会自动生成该权限，需要手动补充，在你的app.py中添加以下代码：

from aws_cdk import aws_iam as iam

# 补充S3调用Lambda的权限
function.add_permission(
    "allow-s3-invoke-lambda",
    principal=iam.ServicePrincipal("s3.amazonaws.com"),
    source_arn=output_s3.bucket_arn,
    action="lambda:InvokeFunction"
)

部署后可前往Lambda控制台「配置」-「权限」-「基于资源的策略」页面确认策略已生效。

2. 确认Lambda入口配置正确 #

你的Lambda函数代码入口为lambda_handler.py中的main函数，需要在CDK定义Lambda时将handler参数设置为lambda_handler.main，如果错误配置为其他入口（比如默认的lambda_function.lambda_handler），会导致函数无法正确接收事件参数，出现空事件输出。

3. 确认S3事件通知配置已生效 #

部署完成后前往S3控制台，找到对应桶的「属性」-「事件通知」板块，确认已存在对应两类事件的通知规则，且目标指向正确的Lambda函数。
如果规则不存在，可先将事件类型替换为全量创建事件临时测试：

output_s3.add_event_notification(
    _s3.EventType.OBJECT_CREATED_ALL,
    notification
)

4. 简化场景测试 #

先通过S3控制台直接上传小于1MB的测试文件，观察Lambda是否能正常收到事件：

• 如果能收到：说明配置正常，可排查CLI上传时是否指定了特殊参数（比如--endpoint-url指向了非官方端点、开启了服务端加密配置不兼容等）
• 如果不能收到：可检查桶是否有其他拦截规则（比如S3访问控制策略、桶日志配置异常等）

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