Я хотел бы знать, существует ли ключ в boto3. Я могу закодировать содержимое ведра и проверить ключ, если он соответствует.
Но это кажется длиннее и излишним. Официальные документы Boto3 явно указывают, как это сделать.
Может быть, мне не хватает очевидного. Может ли кто-нибудь указать мне, как я могу это достичь.
Boto 2 boto.s3.key.Key объект, используемый для использования метода exists, который проверял, существовал ли ключ на S3, выполняя запрос HEAD и просматривая результат, но кажется, что этого больше нет. Вы должны сделать это сами:
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(), а затем обработать там случай ошибки.
Я не большой поклонник использования исключений для потока управления. Это альтернативный подход, который работает в boto3:
import boto3
s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
key = 'dootdoot.jpg'
objs = list(bucket.objects.filter(Prefix=key))
if len(objs) > 0 and objs[0].key == key:
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
В Boto3, если вы проверяете папку (префикс) или файл, используя list_objects. Вы можете использовать существование "Содержание" в ответе dict как проверку того, существует ли объект. Это еще один способ избежать try/except catch, поскольку @EvilPuppetMaster предлагает
import boto3
client = boto3.client('s3')
results = client.list_objects(Bucket='my-bucket', Prefix='dootdoot.jpg')
return 'Contents' in results
Не только 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')
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)
Вы можете использовать S3F, которые по сути являются оболочкой для boto3, которая предоставляет типичные операции в стиле файловой системы:
import s3fs
s3 = s3fs.S3FileSystem()
s3.exists('myfile.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
Попробуйте это просто
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")
Это самый простой метод, который я нашел:
response = s3_client.list_objects_v2(
Bucket=bucket_name,
MaxKeys=1,
Prefix=key_name
)
file_exists = response['KeyCount'] > 0:
Если у вас меньше 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
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. Нам не нужно использовать исключение для этого
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")
Для 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')
Вызывает s3.Client.head_object для обновления атрибутов ресурса ObjectSummary.
Это показывает, что вы можете использовать ObjectSummary вместо Object если вы планируете не использовать get(). Функция load() не извлекает объект, а только сводку.
Если вы ищете ключ, который эквивалентен каталогу, то вам может понадобиться такой подход
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
Это работает для родительского ключа или ключа, который соответствует файлу, или ключа, который не существует. Я попробовал предпочтительный подход выше и не удалось на родительских ключах.
Я заметил, что для того, чтобы botocore.exceptions.ClientError исключение с помощью botocore.exceptions.ClientError нам нужно установить botocore. botocore занимает 36M дискового пространства. Это особенно важно, если мы используем лямбда-функции aws. Вместо этого, если мы просто используем исключение, мы можем пропустить дополнительную библиотеку!
Код выглядит так. Пожалуйста, поделитесь своими мыслями:
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."
Отъезд
bucket.get_key(
key_name,
headers=None,
version_id=None,
response_headers=None,
validate=True
)
Проверьте, существует ли конкретный ключ в ведре. Этот метод использует запрос HEAD для проверки наличия ключа. Возврат: экземпляр объекта Key или None
Вы можете просто вызвать bucket.get_key (имя ключа) и проверить, нет ли возвращенного объекта None.
Вы можете использовать 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 does not exist!")
Здесь ключ - путь, который вы хотите проверить, существует или нет