使用 boto3 检查 s3 中的存储桶中是否存在密钥

Question

我想知道 boto3 中是否存在密钥。我可以循环存储桶内容并检查密钥是否匹配。

但这似乎更长而且有点矫枉过正。 Boto3 官方文档明确说明了如何执行此操作。

可能是我遗漏了显而易见的东西。谁能告诉我如何实现这一目标。

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()，然后在那里处理错误情况。

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

Answer 3

退房

bucket.get_key(
    key_name, 
    headers=None, 
    version_id=None, 
    response_headers=None, 
    validate=True
)

Check to see if a particular key exists within the bucket. This method uses a HEAD request to check for the existence of the key. Returns: An instance of a Key object or None

来自 Boto S3 Docs

您可以调用 bucket.get_key(keyname) 并检查返回的对象是否为 None。

Answer 4

在 Boto3 中，如果您使用 list_objects 检查文件夹（前缀）或文件。您可以在响应字典中使用 'Contents' 的存在性来检查对象是否存在。正如@EvilPuppetMaster 建议的那样，这是避免 try/except 捕获的另一种方法

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

Answer 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

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

Answer 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

Answer 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

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

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

Answer 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

Answer 12

对于boto3，ObjectSummary可用于检查对象是否存在。

Contains the summary of an object stored in an Amazon S3 bucket. This object doesn't contain contain the object's full metadata or any of its contents

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

Calls s3.Client.head_object to update the attributes of the ObjectSummary resource.

这表明如果您不打算使用 get()，则可以使用 ObjectSummary 而不是 Object。 load() 函数不检索对象，它只获取摘要。

Answer 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

这适用于父键或等同于文件的键或不存在的键。我尝试了上面最喜欢的方法，但在父键上失败了。

Answer 14

我注意到，为了使用 botocore.exceptions.ClientError 捕获异常，我们需要安装 botocore。 botocore占用36M磁盘space。如果我们使用 aws lambda 函数，这会产生特别大的影响。取而代之的是，如果我们只使用异常，那么我们可以跳过使用额外的库！

我正在验证文件扩展名为“.csv”
如果桶不存在，这不会抛出异常！
如果桶存在但对象不存在，这不会抛出异常！
如果桶是空的，这会抛出异常！
如果存储桶没有权限，这会抛出异常！

代码如下所示。请分享您的想法：

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)

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

Answer 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."

Answer 17

您可以使用 S3Fs，它本质上是 boto3 的包装器，公开了典型的 file-system 样式操作：

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

Answer 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是你要检查的路径是否存在

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

Answer 20

紧跟话题，有人可以得出哪种方法是检查对象是否存在于 S3 中的最有效方法吗？

我认为 head_object 可能会赢，因为它只检查比实际对象本身更轻的元数据

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

Answer 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

Answer 23

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

Answer 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

does_object_exist 方法使用 s3 客户端的 head_object 方法并检查是否引发了 ClientError。如果错误代码是 404，则返回 False。

使用 boto3 检查 s3 中的存储桶中是否存在密钥

check if a key exists in a bucket in s3 using boto3

python

amazon-s3

boto3