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

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

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


当前回答

假设您只是想检查一个键是否存在(而不是悄悄地覆盖它),首先进行这个检查。也会检查错误:

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

其他回答

假设您只是想检查一个键是否存在(而不是悄悄地覆盖它),首先进行这个检查。也会检查错误:

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

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

不仅是客户端,还有桶:

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

有一种简单的方法可以检查文件是否存在于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")