我想知道boto3中是否存在一个键。我可以循环桶内容并检查键是否匹配。
但这似乎太长了,也太过分了。Boto3官方文档明确说明了如何做到这一点。
也许我忽略了最明显的一点。有人能告诉我怎么做吗?
我想知道boto3中是否存在一个键。我可以循环桶内容并检查键是否匹配。
但这似乎太长了,也太过分了。Boto3官方文档明确说明了如何做到这一点。
也许我忽略了最明显的一点。有人能告诉我怎么做吗?
当前回答
我不太喜欢在控制流中使用异常。这是在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")
其他回答
我发现的最简单的方法(可能也是最有效的)是:
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
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'))
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_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
使用对象。过滤器和检查结果列表是目前为止检查文件是否存在于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)))