如何在Serverless Python Lambda中引用其他栈S3桶作为环境变量与触发源

我来帮你搞定这个需求，直接上可复用的配置和代码示例，都是实战里常用的写法：

1. Serverless配置核心片段（serverless.yml） #

这里假设你其他云栈已经导出了源S3桶的名称和ARN（比如导出键是SourceBucketName和SourceBucketArn），当前栈里定义了目标S3桶：

service: file-conversion-service

provider:
  name: aws
  runtime: python3.11
  region: us-east-1
  # 配置环境变量：把跨栈的源桶名和当前栈的目标桶名传入Lambda
  environment:
    SOURCE_S3_BUCKET: ${cf:other-stack-name.SourceBucketName} # 引用其他栈导出的源桶名
    DESTINATION_S3_BUCKET: !Ref DestinationBucket # 引用当前栈的目标桶

functions:
  file-converter:
    handler: handler.convert_file # 指向Python代码里的handler函数
    # 配置S3触发事件：监听跨栈源桶的文件创建事件
    events:
      - s3:
          bucket: ${cf:other-stack-name.SourceBucketName}
          event: s3:ObjectCreated:*
          rules:
            - prefix: uploads/ # 可选：只监听特定前缀的文件
            - suffix: .txt # 可选：只监听特定后缀的文件
          existing: true # 如果源桶已经存在，必须加这个参数
    # 给Lambda配置必要的权限
    iamRoleStatements:
      - Effect: Allow
        Action:
          - s3:GetObject
        Resource: ${cf:other-stack-name.SourceBucketArn}/* # 允许读取源桶的所有文件
      - Effect: Allow
        Action:
          - s3:PutObject
        Resource: !Sub "${DestinationBucket.Arn}/*" # 允许写入目标桶的所有文件

# 当前栈定义目标S3桶
resources:
  Resources:
    DestinationBucket:
      Type: AWS::S3::Bucket
      Properties:
        BucketName: my-converted-files-bucket-${self:provider.stage} # 按环境区分桶名

2. Python Lambda函数示例（handler.py） #

在代码里直接读取环境变量，不用硬编码桶名，保持配置和代码分离：

import os
import boto3
from io import BytesIO

s3 = boto3.client('s3')
SOURCE_BUCKET = os.environ['SOURCE_S3_BUCKET']
DESTINATION_BUCKET = os.environ['DESTINATION_S3_BUCKET']

def convert_file(event, context):
    # 从S3事件里获取触发的文件信息
    record = event['Records'][0]
    source_key = record['s3']['object']['key']
    
    try:
        # 从源桶下载文件
        response = s3.get_object(Bucket=SOURCE_BUCKET, Key=source_key)
        file_content = response['Body'].read()
        
        # 这里写你的文件转换逻辑，比如把txt转成csv
        converted_content = file_content.decode('utf-8').replace(',', '|').encode('utf-8')
        
        # 生成目标桶的文件键（比如在原键前加converted/前缀）
        destination_key = f"converted/{source_key}"
        
        # 上传到目标桶
        s3.put_object(
            Bucket=DESTINATION_BUCKET,
            Key=destination_key,
            Body=BytesIO(converted_content)
        )
        
        print(f"Successfully converted {source_key} to {destination_key}")
        return {"statusCode": 200, "body": "Conversion completed"}
    except Exception as e:
        print(f"Error processing file {source_key}: {str(e)}")
        raise e

3. 关键注意事项 #

• 跨栈资源引用：确保其他云栈已经通过Outputs导出了源桶的名称和ARN，比如其他栈的serverless.yml里要加：

  resources:
    Outputs:
      SourceBucketName:
        Value: !Ref SourceBucket
        Export:
          Name: SourceBucketName
      SourceBucketArn:
        Value: !GetAtt SourceBucket.Arn
        Export:
          Name: SourceBucketArn
  

• 触发事件权限：如果源桶是手动创建的或者在其他栈，需要确保S3桶有权限向Lambda发送事件通知。如果用上面的existing: true配置，Serverless会尝试自动配置，但如果失败，你可能需要手动在源桶的控制台添加Lambda作为事件目标。
• 环境变量安全：如果桶名包含敏感信息，不用太担心，环境变量在Lambda里是加密存储的（默认用AWS KMS）。

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