使用boto3检查S3存储桶中是否存在一个键。

Question

我想知道在boto3中是否存在一个键。我可以循环遍历存储桶内容并检查键是否匹配。

但这似乎过于冗长和繁琐。Boto3官方文档明确说明了如何做到这一点。

也许我忽略了显而易见的方法。有人能告诉我如何实现这个吗。

- Prabhakar Shanmugam

Answer 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 Nauta

---

感谢您的快速回复，Wander。我只需要boto3的相同内容。 - Prabhakar Shanmugam

对于 boto3，目前最好的方法似乎是调用 head_object 尝试获取密钥的元数据，然后处理结果错误如果它不存在。 - Wander Nauta

@Leonid，你当然可以这样做，但前提是你将其封装在一个函数或方法中，这取决于你。我稍微修改了示例代码，使exists布尔值消失了，并且更清晰（希望如此！）地表明人们应该根据自己的情况进行调整。 - Wander Nauta

-1; doesn't work for me. On boto3 version 1.5.26 I see e.response['Error']['Code'] having a value like "NoSuchKey", not "404". I haven't checked whether this is due to a difference in library versions or a change in the API itself since this answer was written. Either way, in my version of boto3, a shorter approach than checking e.response['Error']['Code'] is to catch only s3.meta.client.exceptions.NoSuchKey in the first place.在我的boto3版本1.5.26上，我看到e.response ['Error'] ['Code']的值类似于“NoSuchKey”，而不是“404”。 我没有检查这是由于库版本的差异还是API本身的更改。 无论哪种方式，在我的boto3版本中，一个比检查e.response['Error']['Code']更短的方法是首先仅捕获s3.meta.client.exceptions.NoSuchKey。 - Mark Amery

如果您正在使用S3“客户端”（而不是“资源”），则请执行s3.head_object(Bucket='my_bucket', Key='my_key')而不是s3.Object(...).load()。 - sam-6174

Answer 2

我找到的最简单的方法（可能也是最有效的）是这样的：

import boto3
import botocore
from botocore.errorfactory import ClientError

s3 = boto3.client('s3')
try:
    s3.head_object(Bucket='bucket_name', Key='file_path')
except botocore.exceptions.ClientError as e:
    if e.response['Error']['Code'] == "404":
        # The key does not exist.
        ...
    elif e.response['Error']['Code'] == 403:
        # Unauthorized, including invalid bucket
        ...
    else:
      # Something else has gone wrong.
      raise 

- o_c

---

注意：如果使用角色或在您的.aws配置文件中拥有密钥，则无需传递aws_access_key_id/aws_secret_access_key等，您只需执行s3 = boto3.client('s3')即可。 - Andy Hayden

我认为添加这个测试会让你更有信心，确信对象确实不存在，而不是其他错误引发了异常——请注意，'e'是ClientError异常实例：if e.response['ResponseMetadata']['HTTPStatusCode'] == 404: - Richard

@Taylor 这是一个没有数据传输的 GET 请求。 - Andy Hayden

ClientError是400的总捕获，而不仅仅是404，因此它不够健壮。 - mickzer

你说得对，更好的方式是接受 S3.Client.exceptions.NoSuchKey。 - Nevenoe

Answer 3

我不太喜欢使用异常来控制流程。这是一种在boto3中可行的替代方法：

import boto3

s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
key = 'dootdoot.jpg'
objs = list(bucket.objects.filter(Prefix=key))
keys = set(o.key for i in objs)
if path_s3 in keys:
    print("Exists!")
else:
    print("Doesn't exist")

- EvilPuppetMaster

---

感谢您的更新，EvilPuppetMaster。不幸的是，上次我检查时我没有列出存储桶访问权限。您的答案非常适合我的问题，所以我已经为您点赞了。但是在那之前我已经将第一个回复标记为答案了。感谢您的帮助。 - Prabhakar Shanmugam

这算不算一种列举请求（价格比获取贵12.5倍）？如果您对1亿个对象执行此操作，那可能会变得有点昂贵...我感觉捕获异常的方法到目前为止是最好的选择。 - Pierre D

列表每个请求可能会比获取单个对象昂贵12.5倍，但是单个列表请求可以返回1亿个对象，而单个获取请求只能返回一个。因此，在您的假设情况下，使用列表获取全部1亿个对象并在本地进行比较比逐个获取100m个对象更便宜。更不用说速度快1000倍，因为您不需要为每个对象进行http往返。 - EvilPuppetMaster

请使用S3客户端的list_objects_v2函数，将MaxKeys设为1。 - Fang Zhang

通过再次使用调试，看起来bucket.objects.filter(Prefix=key).limit(1)不会限制S3实际响应的大小，只是在客户端返回的集合上进行了限制。相反，您应该像@FangZhang建议的那样使用bucket.objects.filter(Prefix=key, MaxKeys=1)。 - Benav

Answer 4

在Boto3中，如果你想要检查一个文件夹（前缀）或者一个文件是否存在，可以使用list_objects方法并检查响应字典中是否存在“Contents”来确定对象是否存在。这是另一种方式来避免try/except语句，正如@EvilPuppetMaster所建议的那样。

import boto3
client = boto3.client('s3')
results = client.list_objects(Bucket='my-bucket', Prefix='dootdoot.jpg')
return 'Contents' in results

- Lucian Thorr

---

在这方面遇到了问题。list_objects("2000") 将返回类似于 "2000-01"、"2000-02" 的键。 - G. Cheng

仅返回1000个对象！https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.list_objects - RoachLord

这是最有效的解决方案，因为它不需要s3:GetObject权限，只需要s3:ListBucket权限。 - Vishrant

Answer 5

这可以检查前缀和键，最多获取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

- Fang Zhang

---

这个答案看起来是众多答案中最好的。唯一的问题是，如果具有键“Key12”的对象存在，则它将返回“True”作为前缀= Key1。要检查确切的键是否存在，最后一行应该修正为类似以下的内容（抱歉代码很丑，我需要在注释中将其放在一行中）：return 'Contents' in response and len(response['Contents']) > 0 and 'Key' in response['Contents'][0] and response['Contents'][0]['Key'] == prefix。 - elshev

Answer 6

假设您只想检查键是否存在（而不是静默地覆盖它），请先进行此检查。这也将检查错误：

import boto3

def key_exists(mykey, mybucket):
    s3_client = boto3.client('s3')
    try:
        response = s3_client.list_objects_v2(Bucket=mybucket, Prefix=mykey)
        for obj in response['Contents']:
            if mykey == obj['Key']:
                return 'exists'
        return False  # no keys match
    except KeyError:
        return False  # no keys found
    except Exception as e:
        # Handle or log other exceptions such as bucket doesn't exist
        return e

key_check = key_exists('someprefix/myfile-abc123', 'my-bucket-name')
if key_check:
    if key_check == 'exists':
        print("key exists!")
    else:
        print(f"S3 ERROR: {e}")
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...

- marvls

---

当我运行此代码时，如果键不存在，将会抛出KeyError: 'Contents'的错误。如果键存在，则正常工作，但也许需要为键缺失情况设置try/except语句块。 - Edward

谢谢Edward，你说得对。我把try/except语句添加到函数中，并添加了一个额外的异常来捕获所有其他S3错误。 - marvls

Answer 7

您可以使用 S3Fs，它本质上是 boto3 的包装器，可公开典型的文件系统样式操作：

import s3fs
s3 = s3fs.S3FileSystem()
s3.exists('myfile.txt')

- VinceP

---

虽然我认为这种方法可行，但问题要求使用boto3完成；在这种情况下，最好不要安装额外的库来解决问题。 - paulkernfeld

此外，s3fs 在技术上是一种将 s3 视为本地目录的挂载机制。除了它的优点之外，在同时读取多个文件时也有许多缺点。 - mandar munagekar

在我的情况下，我正在使用它的S3路径在数据框中读取CSV文件，这需要s3fs。因此，这非常适合我的用例，在读取文件之前检查文件是否存在。 - Yaser Sakkaf

Answer 8

您可以使用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”是你想要检查是否存在的路径

- AshuGG

---

从一个简单的%timeit测试中来看，这似乎是最快的选择。 - Itamar Katz

Answer 9

不仅是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')

- Vitaly Zdanevich

---

您可能并不想获取对象，只是想查看其是否存在。您可以使用像其他示例中一样访问对象的方法，例如 bucket.Object(key).last_modified。 - ryanjdillon

Answer 10

使用objects.filter并检查结果列表是目前在S3存储桶中检查文件是否存在的最快方法。

使用这个简洁的一行代码，当你需要将其放入现有项目而不修改太多代码时，它会变得不那么突兀。

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)))

- nehem

---

我认为这是最好的答案。 - Ramin Melikov