使用LocalStack与Python Boto3配置S3触发Lambda时失败

我在LocalStack环境中尝试设置S3桶的s3:ObjectCreated事件触发AWS Lambda函数，但调用PutBucketNotificationConfiguration操作时出现以下错误：

An error occurred (InvalidArgument) when calling the PutBucketNotificationConfiguration operation: Unable to validate the following destination configurations

我是AWS新手，可能遗漏了某些配置，以下是我的Python代码：

import json

import boto3

print("save script started")
# Initialize a session using LocalStack
session = boto3.Session(
    aws_access_key_id='test',
    aws_secret_access_key='test',
    region_name='eu-west-3',
)

# Create S3 and Lambda clients
s3_client = session.client('s3', endpoint_url='http://localhost:4566')
lambda_client = session.client('lambda', endpoint_url='http://localhost:4566')
iam_client = session.client('iam', endpoint_url='http://localhost:4566')


try:
    policy = iam_client.get_policy(PolicyArn='arn:aws:iam::000000000000:policy/s3-access')
    print("Policy already exists.")
    print(policy)
except iam_client.exceptions.NoSuchEntityException:
    print("Policy does not exist.")
    policy_document = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents",
                "logs:CreateLogGroup",
                "logs:CreateLogStream"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject"],
            "Resource": "arn:aws:s3:::input/*"
        }
    ]
    }

    iam_client.create_policy(
        PolicyName='s3-access',
        PolicyDocument=json.dumps(policy_document)
    )
except:
    print("An error occurred.")

try:
    role = iam_client.get_role(RoleName='execution_role')
    print("Role already exists.")
    print(role)
except iam_client.exceptions.NoSuchEntityException:
    print("Role does not exist.")
    assume_role_policy_document = {
        "Version":"2012-10-17",
        "Statement":[
            {
                "Effect":"Allow",
                "Principal":{"Service":"lambda.amazonaws.com"},
                "Action":"sts:AssumeRole"
            }
        ]
    }
    iam_client.create_role(
        RoleName='execution_role',
        AssumeRolePolicyDocument=json.dumps(assume_role_policy_document)
    )
except:
    print("An error occurred.")


iam_client.attach_role_policy(
    RoleName='execution_role',
    PolicyArn='arn:aws:iam::000000000000:policy/s3-access'
)



try:
    s3_client.head_bucket(Bucket='input')
    print(f"Bucket {'input'} exists.")
except s3_client.exceptions.NoSuchBucket:
    print(f"Bucket {'input'} does not exist.")

s3_client.upload_file('/tmp/my-lambda-function.zip', 'input', 'my-lambda-function.zip')

# Create a Lambda function
try:
    lambda_client.get_function(FunctionName='save')
    print("Function already exists.")
except: 
    lambda_client.create_function(
    FunctionName='save',
    Runtime='python3.10',
    Role='arn:aws:iam::000000000000:role/execution_role',
    Handler='lambda_function.lambda_handler',
    Code={'S3Bucket': 'input', 'S3Key': 'my-lambda-function.zip'},
  )
finally:
    print("Adding permission to Lambda function...")
    lambda_client.add_permission(
        FunctionName='save',
        StatementId='s3-trigger-permission',
        Action='lambda:InvokeFunction',
        Principal='s3.amazonaws.com',
        SourceArn='arn:aws:s3:::input'
    )
    
    print("Adding bucket notification configuration...")
    try:
        s3_client.put_bucket_notification_configuration(
            Bucket='input',
            NotificationConfiguration={
                'LambdaFunctionConfigurations': [
                    {
                        'LambdaFunctionArn': 'arn:aws:lambda:eu-west-3:000000000000:function:save',
                        'Events': ['s3:ObjectCreated:*'],
                    },
                ],
            },
        )
    except Exception as e:
        print(f"An error occurred: {e}")

my-lambda-function.zip已存储在S3桶中。

问题排查与修复方案 #

1. 确保S3桶已创建

LocalStack中S3桶不会自动生成，代码仅检查桶是否存在但未创建，需补充创建逻辑：
修改桶检查的代码块：

try:
    s3_client.head_bucket(Bucket='input')
    print(f"Bucket {'input'} exists.")
except s3_client.exceptions.NoSuchBucket:
    print(f"Bucket {'input'} does not exist. Creating...")
    # LocalStack创建桶需指定区域配置
    s3_client.create_bucket(
        Bucket='input',
        CreateBucketConfiguration={'LocationConstraint': 'eu-west-3'}
    )

2. 完善Lambda权限配置

AWS要求指定SourceArn时最好同时搭配SourceAccount，避免验证失败；同时添加异常处理避免重复添加权限报错：
修改add_permission部分：

print("Adding permission to Lambda function...")
try:
    lambda_client.add_permission(
        FunctionName='save',
        StatementId='s3-trigger-permission',
        Action='lambda:InvokeFunction',
        Principal='s3.amazonaws.com',
        SourceArn='arn:aws:s3:::input',
        SourceAccount='000000000000'
    )
except lambda_client.exceptions.ResourceConflictException:
    print("Permission already exists.")

3. 确保IAM策略同步生效

LocalStack中IAM角色附加策略后可能需要短暂同步，可在创建角色并附加策略后添加延迟（如import time; time.sleep(2)），确保策略生效后再创建Lambda函数。

4. 验证Lambda函数状态

将finally块中的权限配置和通知逻辑调整至Lambda函数创建成功的分支后，避免函数未完全创建时执行后续操作。

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