periksa apakah ada kunci dalam ember di s3 menggunakan boto3

Saya ingin tahu apakah ada kunci di boto3. Saya dapat memutar konten bucket dan memeriksa kunci apakah cocok.

Tapi itu sepertinya lebih lama dan berlebihan. Dokumen resmi Boto3 secara eksplisit menyatakan bagaimana melakukan ini.

Mungkin saya kehilangan yang jelas. Adakah yang bisa menunjukkan kepada saya bagaimana saya bisa mencapai ini.

boto.s3.key.KeyObjek Boto 2 dulu memiliki existsmetode yang memeriksa jika kunci ada pada S3 dengan melakukan permintaan HEAD dan melihat hasilnya, tetapi tampaknya tidak ada lagi. Anda harus melakukannya sendiri:

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() apakah permintaan HEAD untuk satu kunci, yang cepat, bahkan jika objek tersebut besar atau Anda memiliki banyak objek di ember Anda.

Tentu saja, Anda mungkin memeriksa apakah objek itu ada karena Anda berencana menggunakannya. Jika demikian, Anda bisa melupakan load()dan melakukan get()atau download_file()langsung, kemudian menangani kasing kesalahan di sana.

Saya bukan penggemar menggunakan pengecualian untuk aliran kontrol. Ini adalah pendekatan alternatif yang bekerja di 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")

Cara termudah yang saya temukan (dan mungkin yang paling efisien) adalah ini:

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

Di Boto3, jika Anda memeriksa folder (awalan) atau file menggunakan list_objects. Anda dapat menggunakan keberadaan 'Konten' di dict tanggapan sebagai cek untuk apakah objek itu ada. Ini cara lain untuk menghindari try / kecuali tangkapan seperti yang disarankan @EvilPuppetMaster

import boto3
client = boto3.client('s3')
results = client.list_objects(Bucket='my-bucket', Prefix='dootdoot.jpg')
return 'Contents' in results

Tidak hanya clienttetapi bucketjuga:

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

Anda dapat menggunakan S3F , yang pada dasarnya adalah pembungkus di sekitar boto3 yang memaparkan operasi gaya sistem file yang khas:

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

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)

FWIW, berikut adalah fungsi yang sangat sederhana yang saya gunakan

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

Dengan asumsi Anda hanya ingin memeriksa apakah ada kunci (alih-alih menulisnya dengan pelan), lakukan ini terlebih dahulu:

import boto3

def key_exists(mykey, mybucket):
  s3_client = boto3.client('s3')
  response = s3_client.list_objects_v2(Bucket=mybucket, Prefix=mykey)
  if response:
      for obj in response['Contents']:
          if mykey == obj['Key']:
              return True
  return False

if key_exists('someprefix/myfile-abc123', 'my-bucket-name'):
    print("key exists")
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...

Ini bisa memeriksa awalan dan kunci, dan mengambil paling banyak 1 kunci.

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

Coba ini sederhana

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

Jika Anda memiliki kurang dari 1000 dalam direktori atau ember Anda bisa mendapatkan mereka dan setelah memeriksa apakah kunci tersebut dalam set ini:

files_in_dir = {d['Key'].split('/')[-1] for d in s3_client.list_objects_v2(
Bucket='mybucket',
Prefix='my/dir').get('Contents') or []}

Kode seperti itu berfungsi bahkan jika my/dirtidak ada.

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

Untuk boto3, ObjectSummary dapat digunakan untuk memeriksa apakah suatu objek ada.

Berisi ringkasan objek yang disimpan dalam ember Amazon S3. Objek ini tidak mengandung mengandung metadata penuh objek atau isinya

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

Di ObjectSummary.load

Panggilan s3.Client.head_object untuk memperbarui atribut sumber daya ObjectSummary.

Ini menunjukkan bahwa Anda dapat menggunakan ObjectSummarydaripada Objectjika Anda berencana untuk tidak menggunakan get(). The load()fungsi tidak mengambil objek hanya memperoleh ringkasan.

Ini solusi yang cocok untuk saya. Satu peringatan adalah bahwa saya tahu format yang tepat dari kunci sebelumnya, jadi saya hanya mendaftar satu file

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

Anda dapat menggunakan Boto3 untuk ini.

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 doesn't exist!")

Di sini kuncinya adalah jalur yang ingin Anda periksa ada atau tidak

Sangat sederhana dengan get()metode

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

Ada satu cara sederhana yang dengannya kita dapat memeriksa apakah file ada atau tidak di bucket S3. Kami tidak perlu menggunakan pengecualian untuk ini

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

Jika Anda mencari kunci yang setara dengan direktori maka Anda mungkin menginginkan pendekatan ini

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

Ini berfungsi untuk kunci induk atau kunci yang sama dengan file atau kunci yang tidak ada. Saya mencoba pendekatan yang disukai di atas dan gagal pada kunci induk.

Saya perhatikan bahwa hanya untuk menangkap pengecualian menggunakan botocore.exceptions.ClientErrorkita perlu menginstal botocore. botocore membutuhkan 36M ruang disk. Ini sangat berdampak jika kita menggunakan fungsi aws lambda. Di tempat itu jika kita hanya menggunakan pengecualian maka kita bisa melewatkan menggunakan perpustakaan tambahan!

• Saya memvalidasi untuk ekstensi file menjadi '.csv'
• Ini tidak akan memberikan pengecualian jika ember tidak ada!
• Ini tidak akan membuang pengecualian jika bucket ada tetapi objek tidak ada!
• Ini membuang pengecualian jika ember kosong!
• Ini membuang pengecualian jika ember tidak memiliki izin!

Kode terlihat seperti ini. Silakan bagikan pemikiran Anda:

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)

Cukup mengikuti utas, dapatkah seseorang menyimpulkan mana yang merupakan cara paling efisien untuk memeriksa apakah suatu objek ada di S3?

Saya pikir head_object mungkin menang karena hanya memeriksa metadata yang lebih ringan dari objek itu sendiri

Dari https://www.peterbe.com/plog/fastest-way-to-find-out-if-a-file-exists-in-s3 ini ditunjukkan sebagai metode tercepat:

import boto3

boto3_session = boto3.session.Session()
s3_session_client = boto3_session.client("s3")
response = s3_session_client.list_objects_v2(
    Bucket=bc_df_caches_bucket, Prefix=s3_key
)
for obj in response.get("Contents", []):
    if obj["Key"] == s3_key:
        return True
return False

Periksa

bucket.get_key(
    key_name, 
    headers=None, 
    version_id=None, 
    response_headers=None, 
    validate=True
)

Periksa untuk melihat apakah ada kunci tertentu di dalam ember. Metode ini menggunakan permintaan KEPALA untuk memeriksa keberadaan kunci. Pengembalian: Sebuah instance dari objek Kunci atau Tidak Ada

dari Boto S3 Docs

Anda bisa memanggil bucket.get_key (keyname) dan memeriksa apakah objek yang dikembalikan adalah None.