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

但这似乎太长了,也太过分了。Boto3官方文档明确说明了如何做到这一点。

也许我忽略了最明显的一点。有人能告诉我怎么做吗?


当前回答

对于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()函数不检索对象,它只获取摘要。

其他回答

Boto 2的Boto。s3.key。Key对象曾经有一个exists方法,通过执行HEAD请求并查看结果来检查Key是否存在于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请求,这是快速的,即使有问题的对象很大,或者bucket中有很多对象。

当然,您可能会检查对象是否存在,因为您计划使用它。如果是这种情况,您可以忘记load(),直接执行get()或download_file(),然后在那里处理错误情况。

如果你在一个目录或桶中有少于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

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

您可以使用S3Fs,它本质上是boto3的包装器,它公开了典型的文件系统风格操作:

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

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