使用 boto3 检查 s3 的存储桶中是不是存在密钥
Posted
技术标签:
【中文标题】使用 boto3 检查 s3 的存储桶中是不是存在密钥【英文标题】：check if a key exists in a bucket in s3 using boto3使用 boto3 检查 s3 的存储桶中是否存在密钥 【发布时间】：2016-02-23 21:37:02 【问题描述】：我想知道 boto3 中是否存在密钥。我可以循环存储桶的内容并检查密钥是否匹配。
但这似乎更长而且有点矫枉过正。 Boto3 官方文档明确说明了如何执行此操作。
可能是我错过了显而易见的事情。谁能指出我如何做到这一点。
【问题讨论】：
【参考方案1】：Boto 2 的 boto.s3.key.Key 对象曾经有一个 exists 方法，该方法通过执行 HEAD 请求并查看结果来检查 S3 上是否存在密钥，但似乎不再存在。你必须自己做：
import boto3
import botocore
s3 = boto3.resource('s3')
try:
s3.Object('my-bucket', 'dootdoot.jpg').load()
except botocore.exceptions.ClientError as e:
if e.response['Error']['Code'] == "404":
# The object does not exist.
...
else:
# Something else has gone wrong.
raise
else:
# The object does exist.
...
load() 对单个键执行 HEAD 请求，速度很快，即使相关对象很大或者您的存储桶中有很多对象。
当然，您可能正在检查对象是否存在，因为您打算使用它。如果是这种情况，您可以忘记load()，直接执行get() 或download_file()，然后在那里处理错误情况。
【讨论】：
感谢 Wander 的快速回复。我只需要 boto3 一样的。 对于boto3，目前看来最好的办法是调用head_object 来尝试获取密钥的元数据，然后处理结果错误（如果它不存在） .
哦，上面的head_bucket 建议仅适用于存储桶，不适用于对象。我撤回我的建议。 :-)
-1;对我不起作用。在 boto3 版本 1.5.26 上，我看到 e.response['Error']['Code'] 的值类似于 "NoSuchKey"，而不是 "404"。自编写此答案以来，我尚未检查这是否是由于库版本的差异或 API 本身的更改。无论哪种方式，在我的 boto3 版本中，比检查 e.response['Error']['Code'] 更短的方法是首先只捕获 s3.meta.client.exceptions.NoSuchKey。
如果您使用的是 s3 client（而不是 resource），则使用 s3.head_object(Bucket='my_bucket', Key='my_key') 而不是 s3.Object(...).load()【参考方案2】：
我不太喜欢在控制流中使用异常。这是一种适用于 boto3 的替代方法：
import boto3
s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
key = 'dootdoot.jpg'
objs = list(bucket.objects.filter(Prefix=key))
if any([w.key == path_s3 for w in objs]):
print("Exists!")
else:
print("Doesn't exist")
【讨论】：
感谢 EvilPuppetMaster 的更新。不幸的是，当我上次检查时，我没有列表存储桶访问权限。你的回答很适合我的问题，所以我投了你的票。但我早就将第一个回复标记为答案。感谢您的帮助。 这算不算一个listing请求（比get贵12.5倍）？如果你对 1 亿个对象执行此操作，那可能会有点贵......我觉得捕获异常方法是迄今为止最好的。 List 可能是每个请求的 12.5 倍，但单个请求也可以返回 1 亿个对象，而单个 get 只能返回一个。因此，在您的假设情况下，使用列表获取所有 1 亿个然后在本地进行比较比单独获取 1 亿个更便宜。更不用说快 1000 倍了，因为您不需要为每个对象都进行 http 往返。 @user3186866 这是因为 S3 实际上没有“文件夹”。所有对象都作为文件存在于它们的给定路径中。文件夹是帮助我们组织和理解存储结构的工具，但实际上，S3 存储桶就是这样的存储桶。 再次使用调试运行后，bucket.objects.filter(Prefix=key).limit(1) 似乎并没有限制来自 S3 的实际响应，只限制了客户端返回的集合。相反，你应该bucket.objects.filter(Prefix=key, MaxKeys=1) 就像上面的@FangZhang 建议的那样【参考方案3】：
退房
bucket.get_key(
key_name,
headers=None,
version_id=None,
response_headers=None,
validate=True
)
检查存储桶中是否存在特定键。这种方法 使用 HEAD 请求检查密钥是否存在。回报：一个 Key 对象或 None 的实例
来自Boto S3 Docs
您可以调用 bucket.get_key(keyname) 并检查返回的对象是否为 None。
【讨论】：
按照 OP 的要求，这不适用于 boto3 AWS boto 库有两个版本。此答案不适用于问题要求的版本。 这肯定不是 OP 的正确答案，但它对我有帮助，因为我需要使用 boto v2。这就是我删除反对票的原因。【参考方案4】：在 Boto3 中，如果您使用 list_objects 检查文件夹（前缀）或文件。您可以使用响应字典中是否存在“内容”来检查对象是否存在。正如@EvilPuppetMaster 建议的那样，这是另一种避免 try/except 捕获的方法
import boto3
client = boto3.client('s3')
results = client.list_objects(Bucket='my-bucket', Prefix='dootdoot.jpg')
return 'Contents' in results
【讨论】：
在这方面有问题。 list_objects("2000") 将返回诸如“2000-01”、“2000-02”之类的键 这只会返回最多 1000 个对象！ boto3.amazonaws.com/v1/documentation/api/latest/reference/… 这是最有效的解决方案，因为它不需要s3:GetObject 权限，只需要s3:ListBucket 权限【参考方案5】：
我发现的最简单的方法（可能也是最有效的）是这样的：
import boto3
from botocore.errorfactory import ClientError
s3 = boto3.client('s3')
try:
s3.head_object(Bucket='bucket_name', Key='file_path')
except ClientError:
# Not found
pass
【讨论】：
注意：您不必传递 aws_access_key_id/aws_secret_access_key 等。如果使用角色或者您的 .aws 配置中有密钥，您只需执行s3 = boto3.client('s3')
我认为添加此测试可以让您更有信心该对象确实不存在，而不是引发异常的其他错误 - 请注意，'e' 是 ClientError 异常实例：if e.response['ResponseMetadata']['HTTPStatusCode'] == 404:
@Taylor 这是一个获取请求，但没有数据传输。
ClientError 是 400 的全部，而不仅仅是 404，因此它不可靠。
@mickzer 你是对的。最好排除 S3.Client.exceptions.NoSuchKey。【参考方案6】：
不仅client，还有bucket：
import boto3
import botocore
bucket = boto3.resource('s3', region_name='eu-west-1').Bucket('my-bucket')
try:
bucket.Object('my-file').get()
except botocore.exceptions.ClientError as ex:
if ex.response['Error']['Code'] == 'NoSuchKey':
print('NoSuchKey')
【讨论】：
您可能不想获取该对象，但只需查看它是否存在即可。您可以像这里的其他示例一样使用指向对象的方法，例如bucket.Object(key).last_modified。【参考方案7】：
如果您的目录或存储桶中少于 1000 个，您可以获取它们的集合，然后检查此集合中是否存在这样的键：
files_in_dir = d['Key'].split('/')[-1] for d in s3_client.list_objects_v2(
Bucket='mybucket',
Prefix='my/dir').get('Contents') or []
即使my/dir 不存在，这样的代码也可以工作。
http://boto3.readthedocs.io/en/latest/reference/services/s3.html#S3.Client.list_objects_v2
【讨论】：
【参考方案8】：S3_REGION="eu-central-1"
bucket="mybucket1"
name="objectname"
import boto3
from botocore.client import Config
client = boto3.client('s3',region_name=S3_REGION,config=Config(signature_version='s3v4'))
list = client.list_objects_v2(Bucket=bucket,Prefix=name)
for obj in list.get('Contents', []):
if obj['Key'] == name: return True
return False
【讨论】：
【参考方案9】：我们可以通过一种简单的方法检查 S3 存储桶中是否存在文件。我们不需要为此使用异常
sesssion = boto3.Session(aws_access_key_id, aws_secret_access_key)
s3 = session.client('s3')
object_name = 'filename'
bucket = 'bucketname'
obj_status = s3.list_objects(Bucket = bucket, Prefix = object_name)
if obj_status.get('Contents'):
print("File exists")
else:
print("File does not exists")
【讨论】：
如果存储桶中存在以object_name 开头的文件，这将是不正确的。例如。如果您检查 my_file.txt，my_file.txt.oldversion 将返回误报。对大多数人来说有点极端，但对于像“文件是否存在”这样广泛的问题，您可能会在整个应用程序中使用它可能值得考虑。【参考方案10】：
import boto3
client = boto3.client('s3')
s3_key = 'Your file without bucket name e.g. abc/bcd.txt'
bucket = 'your bucket name'
content = client.head_object(Bucket=bucket,Key=s3_key)
if content.get('ResponseMetadata',None) is not None:
print "File exists - s3://%s/%s " %(bucket,s3_key)
else:
print "File does not exist - s3://%s/%s " %(bucket,s3_key)
【讨论】：
我喜欢这个答案，但如果文件不存在，它就不起作用，它只会抛出一个错误，然后你会被困在做与某些相同的事情其他答案。【参考方案11】：FWIW，这是我正在使用的非常简单的功能
import boto3
def get_resource(config: dict=):
"""Loads the s3 resource.
Expects AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY to be in the environment
or in a config dictionary.
Looks in the environment first."""
s3 = boto3.resource('s3',
aws_access_key_id=os.environ.get(
"AWS_ACCESS_KEY_ID", config.get("AWS_ACCESS_KEY_ID")),
aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", config.get("AWS_SECRET_ACCESS_KEY")))
return s3
def get_bucket(s3, s3_uri: str):
"""Get the bucket from the resource.
A thin wrapper, use with caution.
Example usage:
>> bucket = get_bucket(get_resource(), s3_uri_prod)"""
return s3.Bucket(s3_uri)
def isfile_s3(bucket, key: str) -> bool:
"""Returns T/F whether the file exists."""
objs = list(bucket.objects.filter(Prefix=key))
return len(objs) == 1 and objs[0].key == key
def isdir_s3(bucket, key: str) -> bool:
"""Returns T/F whether the directory exists."""
objs = list(bucket.objects.filter(Prefix=key))
return len(objs) > 1
【讨论】：
这是我看到的唯一一个解决了与“文件”相比检查“文件夹”是否存在的响应。这对于需要知道特定文件夹是否存在而不是文件夹中的特定文件的例程来说非常重要。 虽然这是一个谨慎的答案，但它仅在用户理解文件夹的概念在这种情况下具有误导性时才有用。一个空的“文件夹”可以存在于存储桶内的 S3 中，如果是这样，isdir_s3 将返回 False 我花了几分钟时间来解决这个问题我正在考虑编辑答案，就好像表达式更改为 >0 你会得到你期待的结果【参考方案12】：对于boto3，ObjectSummary可以用来检查对象是否存在。
包含存储在 Amazon S3 存储桶中的对象的摘要。此对象不包含该对象的完整元数据或其任何内容
import boto3
from botocore.errorfactory import ClientError
def path_exists(path, bucket_name):
"""Check to see if an object exists on S3"""
s3 = boto3.resource('s3')
try:
s3.ObjectSummary(bucket_name=bucket_name, key=path).load()
except ClientError as e:
if e.response['Error']['Code'] == "404":
return False
else:
raise e
return True
path_exists('path/to/file.html')
在ObjectSummary.load
调用 s3.Client.head_object 更新 ObjectSummary 资源的属性。
这表明如果您打算不使用get()，则可以使用ObjectSummary 而不是Object。 load() 函数不检索它只获取摘要的对象。
【讨论】：
【参考方案13】：如果您寻求与目录等效的密钥，那么您可能需要这种方法
session = boto3.session.Session()
resource = session.resource("s3")
bucket = resource.Bucket('mybucket')
key = 'dir-like-or-file-like-key'
objects = [o for o in bucket.objects.filter(Prefix=key).limit(1)]
has_key = len(objects) > 0
这适用于父键或等同于文件的键或不存在的键。我尝试了上面最喜欢的方法，但在父键上失败了。
【讨论】：
【参考方案14】：我注意到，为了使用botocore.exceptions.ClientError 捕获异常，我们需要安装 botocore。 botocore 占用了 36M 的磁盘空间。如果我们使用 aws lambda 函数，这尤其会产生影响。取而代之的是，如果我们只使用异常，那么我们可以跳过使用额外的库！
代码如下所示。请分享您的想法：
import boto3
import traceback
def download4mS3(s3bucket, s3Path, localPath):
s3 = boto3.resource('s3')
print('Looking for the csv data file ending with .csv in bucket: ' + s3bucket + ' path: ' + s3Path)
if s3Path.endswith('.csv') and s3Path != '':
try:
s3.Bucket(s3bucket).download_file(s3Path, localPath)
except Exception as e:
print(e)
print(traceback.format_exc())
if e.response['Error']['Code'] == "404":
print("Downloading the file from: [", s3Path, "] failed")
exit(12)
else:
raise
print("Downloading the file from: [", s3Path, "] succeeded")
else:
print("csv file not found in in : [", s3Path, "]")
exit(12)
【讨论】：
AWS 说 python 运行时预装了 boto3：docs.aws.amazon.com/lambda/latest/dg/lambda-runtimes.html【参考方案15】：试试这个简单的
import boto3
s3 = boto3.resource('s3')
bucket = s3.Bucket('mybucket_name') # just Bucket name
file_name = 'A/B/filename.txt' # full file path
obj = list(bucket.objects.filter(Prefix=file_name))
if len(obj) > 0:
print("Exists")
else:
print("Not Exists")
【讨论】：
【参考方案16】：这是一个适合我的解决方案。一个警告是我提前知道密钥的确切格式，所以我只列出单个文件
import boto3
# The s3 base class to interact with S3
class S3(object):
def __init__(self):
self.s3_client = boto3.client('s3')
def check_if_object_exists(self, s3_bucket, s3_key):
response = self.s3_client.list_objects(
Bucket = s3_bucket,
Prefix = s3_key
)
if 'ETag' in str(response):
return True
else:
return False
if __name__ == '__main__':
s3 = S3()
if s3.check_if_object_exists(bucket, key):
print "Found S3 object."
else:
print "No object found."
【讨论】：
【参考方案17】：您可以使用S3Fs，它本质上是对 boto3 的封装，它公开了典型的文件系统样式操作：
import s3fs
s3 = s3fs.S3FileSystem()
s3.exists('myfile.txt')
【讨论】：
虽然我认为这可行，但问题询问如何使用 boto3 执行此操作；在这种情况下，无需安装额外的库即可解决问题。 此外，s3fs 在技术上是一种将 s3 视为本地目录的挂载机制。除了优点之外，它在同时读取多个文件时也有许多缺点。【参考方案18】：您可以为此使用 Boto3。
import boto3
s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
objs = list(bucket.objects.filter(Prefix=key))
if(len(objs)>0):
print("key exists!!")
else:
print("key doesn't exist!")
这里的key是你要检查的路径是否存在
【讨论】：
从简单的%timeit 测试来看，这似乎是最快的选择【参考方案19】：
get() 方法真的很简单
import botocore
from boto3.session import Session
session = Session(aws_access_key_id='AWS_ACCESS_KEY',
aws_secret_access_key='AWS_SECRET_ACCESS_KEY')
s3 = session.resource('s3')
bucket_s3 = s3.Bucket('bucket_name')
def not_exist(file_key):
try:
file_details = bucket_s3.Object(file_key).get()
# print(file_details) # This line prints the file details
return False
except botocore.exceptions.ClientError as e:
if e.response['Error']['Code'] == "NoSuchKey": # or you can check with e.reponse['HTTPStatusCode'] == '404'
return True
return False # For any other error it's hard to determine whether it exists or not. so based on the requirement feel free to change it to True/ False / raise Exception
print(not_exist('hello_world.txt'))
【讨论】：
不健壮，可能由于多种原因引发异常，例如HTTP 500，此代码将假定为 404。 但是我们需要有关文件是否可访问的信息。它存在且无法访问，则相当于不存在。对吗？ @mickzer 现在检查更改。 回复您之前的评论，不，HTTP 500 上的行为可能是重试，401/403 修复身份验证等。检查实际错误代码很重要。跨度> 【参考方案20】：按照这个线程，有人可以得出结论，哪一种是检查 S3 中是否存在对象的最有效方法？
我认为 head_object 可能会获胜，因为它只是检查比实际对象本身更轻的元数据
【讨论】：
是的，head_object 是最快的方法——这也是s3.Object('my-bucket', 'dootdoot.jpg').load() 在后台检查对象是否存在的方式。看这个方法失败时的报错信息就可以看到。【参考方案21】：
假设您只想检查密钥是否存在（而不是悄悄地覆盖它），请先进行以下检查：
import boto3
def key_exists(mykey, mybucket):
s3_client = boto3.client('s3')
response = s3_client.list_objects_v2(Bucket=mybucket, Prefix=mykey)
if response:
for obj in response['Contents']:
if mykey == obj['Key']:
return True
return False
if key_exists('someprefix/myfile-abc123', 'my-bucket-name'):
print("key exists")
else:
print("safe to put new bucket object")
# try:
# resp = s3_client.put_object(Body="Your string or file-like object",
# Bucket=mybucket,Key=mykey)
# ...check resp success and ClientError exception for errors...
【讨论】：
【参考方案22】：这可以同时检查前缀和键，并且最多获取 1 个键。
def prefix_exits(bucket, prefix):
s3_client = boto3.client('s3')
res = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=1)
return 'Contents' in res
【讨论】：
【参考方案23】：使用objects.filter 并检查结果列表是迄今为止检查文件是否存在于 S3 存储桶中的最快方法。 .
使用这种简洁的 oneliner，当您必须将其放入现有项目中而不修改大部分代码时，它会减少干扰。
s3_file_exists = lambda filename: bool(list(bucket.objects.filter(Prefix=filename)))
上面的函数假设bucket变量已经被声明了。
您可以扩展 lambda 以支持其他参数，例如
s3_file_exists = lambda filename, bucket: bool(list(bucket.objects.filter(Prefix=filename)))
【讨论】：
我认为这是最好的答案。【参考方案24】：您可以使用 awswrangler 在 1 行中完成。
awswrangler.s3.does_object_exist(path_of_object_to_check)
https://aws-data-wrangler.readthedocs.io/en/stable/stubs/awswrangler.s3.does_object_exist.html
dos_object_exist 方法使用 s3 客户端的 head_object 方法并检查是否引发了 ClientError。如果错误代码是 404 则返回 False。
【讨论】：
以上是关于使用 boto3 检查 s3 的存储桶中是不是存在密钥的主要内容，如果未能解决你的问题，请参考以下文章
使用 boto3 lib 和 AWS Lambda 从 S3 存储桶中的压缩文件中获取数据流