使用 Lambda python 将 S3 对象从一个账户复制到另一个账户答案

【问题标题】：Copying S3 objects from one account to other using Lambda python使用 Lambda python 将 S3 对象从一个账户复制到另一个账户
【发布时间】：2021-10-11 18:03:11
【问题描述】：

我正在使用 boto3 将文件从 s3 存储桶从一个帐户复制到另一个帐户。我需要类似 aws s3 sync 的功能。请看我的代码。我的公司已决定从其他 S3 存储桶（源帐户）“拉取”。请不要建议复制、S3 批处理、S3 触发 Lambda 等。我们已经完成了所有这些选项，我的管理层不想在源端进行任何配置。您能否查看此代码并让我知道此代码是否适用于数千个对象。源桶有近 10000 个对象。我们将在目标账户中创建这个 lambda 函数，并创建一个 cloudwatch 事件来每天触发一次 lambda。

我正在检查 ETag，以便在触发此功能时复制修改的文件。

编辑：我简化了我的代码只是为了查看分页工作。如果我不添加 client.copy()，它就可以工作。如果我在读取 3,4 个对象后将这一行添加到 for 循环中，则会抛出“errorMessage”：“2021-08-07T15:29:07.827Z 82757747-7b72-4f29-ae9f-22e95f969d6c 任务在 3.00 秒后超时”。请指教。请注意，我的源存储桶中的“test/”文件夹包含大约 1100 个对象。

import os
import logging
import botocore

logger = logging.getLogger()
logger.setLevel(os.getenv('debug_level', 'INFO'))

client = boto3.client('s3')
def handler(event, context):
    main(event, logger)

def main(event, logger):
    try:
        SOURCE_BUCKET = os.environ.get('SRC_BUCKET')
        DEST_BUCKET = os.environ.get('DST_BUCKET')
        REGION = os.environ.get('REGION')
        prefix = 'test/'
        # Create a reusable Paginator
        paginator = client.get_paginator('list_objects_v2')
        print ('after paginator')
        # Create a PageIterator from the Paginator
        page_iterator = paginator.paginate(Bucket=SOURCE_BUCKET,Prefix = prefix)
        print ('after page iterator')
        index = 0
        for page in page_iterator:
            for obj in page['Contents']:
                index += 1
                print ("I am looking for {} in the source bucket".format(obj['ETag']))
                copy_source = {'Bucket': SOURCE_BUCKET, 'Key': obj['Key']}
                client.copy(copy_source, DEST_BUCKET, obj['Key'])
        logger.info("number of objects copied {}:".format(index))
    except botocore.exceptions.ClientError as e:
        raise

【问题讨论】：

两个桶中超过 1000 个对象可能有问题，因为这是 list_objects_v2() 返回的最大对象数。您应该对结果进行分页，或使用paginator。 if not excluded_dir in obj['Key'] 可能会导致字符串的部分匹配出现问题——最好与 .startswith() 进行比较而不是 in。
另外，如果一个对象以不同的名称被复制，它不会被重新复制。例如，如果 object1 也作为 object2 存在，则只会复制其中一个对象。
谢谢约翰。我会尝试分页器。如果可能的话，你能指出我应该在哪里更改我的代码。我不太明白你的第二条评论。
@JohnRotenstein，我正在我的开发帐户中测试这段代码，只有几个文件，有没有办法用 1000 个对象测试这段代码？
假设您有一个名为 s3copy/ 的排除目录。如果对象存储在foo/bar/s3copy/something/ 中，则if not excluded_dir 行将停止复制对象，因为该目录名称在路径中的elsewhere 中，不一定在路径的开头。

标签： amazon-web-services amazon-s3 aws-lambda

【解决方案1】：

如果我将 Lambda 超时时间增加到 15 分钟并将内存增加到 512MB，则此版本可以正常工作。这会在复制之前检查源对象是否已存在于目标中。

import boto3
import os
import logging
import botocore
from botocore.client import Config

logger = logging.getLogger()
logger.setLevel(os.getenv('debug_level', 'INFO'))

config = Config(connect_timeout=5, retries={'max_attempts': 0})
client = boto3.client('s3', config=config)

#client = boto3.client('s3')
def handler(event, context):
    main(event, logger)

def main(event, logger):
    try:
        DEST_BUCKET = os.environ.get('DST_BUCKET')
        SOURCE_BUCKET = os.environ.get('SRC_BUCKET')
        REGION = os.environ.get('REGION')
        prefix = ''
        # Create a reusable Paginator
        paginator = client.get_paginator('list_objects_v2')
        print ('after paginator')

        # Create a PageIterator from the Paginator
        page_iterator_src = paginator.paginate(Bucket=SOURCE_BUCKET,Prefix = prefix)
        page_iterator_dest = paginator.paginate(Bucket=DEST_BUCKET,Prefix = prefix)
        print ('after page iterator')
        index = 0
        for page_source in page_iterator_src:
            for obj_src in page_source['Contents']:
                flag = "FALSE"
                for page_dest in page_iterator_dest:
                    for obj_dest in page_dest['Contents']:
                        # checks if source ETag already exists in destination
                        if obj_src['ETag'] in obj_dest['ETag']:
                            flag = "TRUE"
                            break
                    if flag == "TRUE":
                        break
                if flag != "TRUE":
                    index += 1
                    client.copy_object(Bucket=DEST_BUCKET, CopySource={'Bucket': SOURCE_BUCKET, 'Key': obj_src['Key']}, Key=obj_src['Key'],)
                    print ("source ETag {} and destination ETag {}".format(obj_src['ETag'],obj_dest['ETag']))
                    print ("source Key {} and destination Key {}".format(obj_src['Key'],obj_dest['Key']))
        print ("Number of objects copied{}".format(index))
        logger.info("number of objects copied {}:".format(index))
    except botocore.exceptions.ClientError as e:
        raise

【讨论】：