我想知道boto3中是否存在一个键。我可以循环桶内容并检查键是否匹配。
但这似乎太长了,也太过分了。Boto3官方文档明确说明了如何做到这一点。
也许我忽略了最明显的一点。有人能告诉我怎么做吗?
我想知道boto3中是否存在一个键。我可以循环桶内容并检查键是否匹配。
但这似乎太长了,也太过分了。Boto3官方文档明确说明了如何做到这一点。
也许我忽略了最明显的一点。有人能告诉我怎么做吗?
当前回答
我注意到,为了使用botocore.exceptions. clienterror捕获异常,我们需要安装botocore。botocore占用36M的磁盘空间。如果我们使用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)
其他回答
这里有一个对我有用的解决办法。需要注意的是,我事先知道密钥的确切格式,所以我只列出单个文件
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."
试试这个简单的方法
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")
有一种简单的方法可以检查文件是否存在于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")
看看
bucket.get_key(
key_name,
headers=None,
version_id=None,
response_headers=None,
validate=True
)
检查bucket中是否存在特定的键。这个方法 使用HEAD请求来检查密钥是否存在。返回: Key对象或None的实例
从Boto S3文档
你可以调用bucket.get_key(keyname)并检查返回的对象是否为None。
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