boto3 - 获取仅在 S3 中上个月上传的文件答案

【问题标题】：boto3 - Getting files only uploaded in the past month in S3boto3 - 获取仅在 S3 中上个月上传的文件
【发布时间】：2022-01-24 04:54:14
【问题描述】：

我正在编写一个 python3 lambda 函数，该函数需要返回从函数运行开始的过去 30 天内上传到 S3 存储桶的所有文件。

我应该如何处理这个问题？理想情况下，我只想遍历过去 30 天的文件，仅此而已 - 我正在遍历的 S3 存储桶中有成千上万个文件，可能每月最多更新/上传 100 个文件。必须遍历每个文件并像这样比较日期是非常低效的。 AWS API 网关也有 29 秒的时间限制。

任何帮助将不胜感激。谢谢！

【问题讨论】：

看看我的回答：stackoverflow.com/a/59146547/9931092
这不是您问题的答案，但我可能建议的一件事是将您的文件按日期分隔保存在您的存储桶中。当我有包含大量文件的存储桶时，我倾向于使用年/月/日/前缀进行保存。以后可能会对你有所帮助
感谢@RafaelMarques 不幸的是，我无权更改任何文件的名称。存储桶由我合作的另一个团队管理。

标签： python-3.x amazon-web-services amazon-s3 aws-lambda boto3

【解决方案1】：

您需要遍历对象列表（示例代码：List s3 buckets with its size in csv format）并比较 Python 代码中的日期（示例代码：Get day old filepaths from s3 bucket）。

列出对象时没有过滤器（除了前缀）。

另一种方法是使用Amazon S3 Inventory，它可以提供列出存储桶内容的每日 CSV 文件。您可以解析该 CSV 而不是列表对象。

一个更极端的选择是保留一个单独的对象数据库，每当添加/删除对象时都需要更新该数据库。这可以通过触发 AWS Lambda 函数的 Amazon S3 事件来完成。不过工作量很大。

【讨论】：

每个单独的上传都会导致调用 Lambda 函数。 event 记录将包含存储桶名称和触发事件的对象的键。
听起来很棒，非常感谢您的快速回复！对不起，我删除了我的评论，因为我想重写它，但不小心删除了它而不是编辑它。
嗨约翰，所以我尝试为我的 lambda 创建 S3 触发器，但我收到错误：“创建触发器时发生错误：配置定义不明确。如果两个规则中不能有重叠的后缀相同事件类型的前缀重叠”。我认为这是因为已经有另一个 lambda 由 S3 的更新触发（这个由特定后缀的上传触发，而对我来说，我需要所有上传的触发器）。不幸的是，我的团队没有授权我触摸其他 lambda。您对此有什么建议吗？
Amazon S3 中的一个操作只能触发一个事件。一种选择是让 S3 向 SNS 主题发送消息，然后该主题可以触发多个 Lambda 函数。但是，如果您的“触发对象”仅部分重叠其他团队的需求，这可能会很困难。我建议使用 Python 代码或 Amazon S3 Inventory。

【解决方案2】：

我不能给你一个 100% 的答案，因为你已经询问了上传日期，但是如果你可以接受“最后修改”的值，这个代码 sn-p 应该可以完成这项工作：

import boto3
import datetime

paginator = boto3.resource('s3').meta.client.get_paginator('list_objects')
date = datetime.datetime.now() - datetime.timedelta(30)
filtered_files = (page['Key'] for page in paginator.paginate(Bucket="bucketname").search(f"Contents[?to_string(LastModified)>='\"{date}\"']"))

我使用JMESPath进行过滤

【讨论】：

【解决方案3】：

从建筑师的角度来看

瓶颈在于是否可以在 30 秒内迭代所有对象。如果本机文件太多，您可以使用更多选项：

创建由S3:PutObject 事件触发的aws lambda 函数，并将S3 密钥和last_modified_at 信息存储到Dynamodb（AWS Key Value NoSQL 数据库）中。然后，您可以轻松地使用 Dynamodb 过滤 S3 密钥并相应地检索这些 S3 对象。
创建由S3:PutObject 事件触发的 aws lambda 函数，并将文件移动到分区的 S3 Key 架构位置，例如 s3://bucket/datalake/year=${year}/month=${month}/day=${day}/your-file.csv。然后，您可以轻松地使用分区信息来定位对象的子集，这符合 30 秒的硬限制。

从编程角度来看

这里的代码 sn-p 解决了你的问题使用这个库s3pathlib：

from datetime import datetime, timedelta
from s3pathlib import S3path

# define a folder
p_dir = S3Path("bucket/my-folder/")

# find one month ago datetime
now = datetime.utcnow()
one_month_ago = now - timedelta(days=30)

# filter by last modified
for p in p_bucket.iter_objects().filter(
    # any Filterable Attribute can be used for filtering
    S3Path.last_modified_at >= one_month_ago 
)：
    # do whatever you like
    print(p.console_url) # click link to open it in console, inspect

如果您想使用其他S3Path 属性进行过滤，并使用其他比较器，甚至定义您的自定义过滤器，您可以按照此document

【讨论】：