一、背景与需求
在实际业务中，经常会遇到这样一个需求：
当文件上传到 S3 后，自动进行处理
本案例实现的目标是：
当文件上传到源 S3 Bucket 后
自动触发处理流程
`import json
import boto3
from datetime import datetime
client = boto3.client('s3')
dynamodb_client = boto3.client('dynamodb')
def lambda_handler(event, context):
# TODO implement
source_bucket_name = "source_bucket_name"
target_bucket_name = "target_bucket_name"
file_time = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")
for record in event['Records']:
body = json.loads(record['body'])
message = json.loads(body['Message'])
print(message)
print(f"message:{message}")
s3_key = message['Records'][0]['s3']['object']['key']
file_size = message['Records'][0]['s3']['object']['size']
print(f"s3_key:{s3_key},file_size:{file_size}")
response = client.copy_object(
Bucket=target_bucket_name,
CopySource=f'/{source_bucket_name}/{s3_key}',
Key=s3_key,
)
dynamodb_response = dynamodb_client.put_item(
Item={
'sync_id': {
'S': s3_key,
},
'file_name': {
'S': s3_key,
},
'source_bucket': {
'S': source_bucket_name,
},
'target_bucket': {
'S': target_bucket_name,
},
'file_size': {
'N': str(file_size),
},
'sync_time': {
'S': file_time,
},
'status': {
'S': 'SUCCESS',
},
},
TableName='TableName',
)
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}
`