vérifier si une clé existe dans un compartiment en s3 en utilisant boto3

Je voudrais savoir si une clé existe dans boto3. Je peux boucler le contenu du seau et vérifier la clé si elle correspond.

Mais cela semble plus long et exagéré. Les documents officiels de Boto3 indiquent explicitement comment procéder.

Peut-être que je manque l'évidence. Quelqu'un peut-il m'indiquer comment je peux y parvenir.

L' boto.s3.key.Keyobjet de Boto 2 avait une existsméthode qui vérifiait si la clé existait sur S3 en faisant une requête HEAD et en regardant le résultat, mais il semble que cela n'existe plus. Vous devez le faire vous-même:

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() fait une requête HEAD pour une seule clé, ce qui est rapide, même si l'objet en question est volumineux ou si vous avez de nombreux objets dans votre compartiment.

Bien sûr, vous pouvez vérifier si l'objet existe parce que vous prévoyez de l'utiliser. Si tel est le cas, vous pouvez simplement oublier le load()et faire un get()ou download_file()directement, puis gérer le cas d'erreur ici.

Je ne suis pas un grand fan de l'utilisation d'exceptions pour le flux de contrôle. C'est une approche alternative qui fonctionne dans 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")

Le moyen le plus simple que j'ai trouvé (et probablement le plus efficace) est le suivant:

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

Dans Boto3, si vous recherchez un dossier (préfixe) ou un fichier à l'aide de list_objects. Vous pouvez utiliser l'existence de «Contenu» dans le dict de la réponse pour vérifier si l'objet existe. C'est une autre façon d'éviter les captures try / except comme le suggère @EvilPuppetMaster

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

Non seulement clientmais bucketaussi:

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

Vous pouvez utiliser S3F , qui est essentiellement un wrapper autour de boto3 qui expose les opérations typiques de style de système de fichiers:

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, voici les fonctions très simples que j'utilise

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

En supposant que vous vouliez simplement vérifier si une clé existe (au lieu de l'écraser tranquillement), faites d'abord cette vérification:

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

Cela peut vérifier à la fois le préfixe et la clé, et récupère au plus 1 clé.

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

Essayez ceci simple

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

Si vous en avez moins de 1000 dans un répertoire ou un compartiment, vous pouvez en obtenir un ensemble et après vérifier si une telle clé dans cet ensemble:

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

Un tel code fonctionne même s'il my/dirn'existe pas.

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

Pour boto3, ObjectSummary peut être utilisé pour vérifier si un objet existe.

Contient le résumé d'un objet stocké dans un compartiment Amazon S3. Cet objet ne contient pas les métadonnées complètes de l'objet ni aucun de ses contenus

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

Dans ObjectSummary.load

Appelle s3.Client.head_object pour mettre à jour les attributs de la ressource ObjectSummary.

Cela montre que vous pouvez utiliser ObjectSummaryau lieu de Objectsi vous prévoyez de ne pas utiliser get(). La load()fonction ne récupère pas l'objet, elle obtient uniquement le résumé.

Voici une solution qui fonctionne pour moi. Une mise en garde est que je connais le format exact de la clé à l'avance, donc je ne liste que le fichier unique

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

vous pouvez utiliser Boto3 pour cela.

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

Ici, la clé est le chemin que vous souhaitez vérifier existe ou non

C'est vraiment simple avec la get()méthode

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

Il existe un moyen simple de vérifier si le fichier existe ou non dans le compartiment S3. Nous n'avons pas besoin d'utiliser une exception pour cela

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

Si vous recherchez une clé équivalente à un répertoire, vous voudrez peut-être cette approche

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

Cela fonctionne pour une clé parente ou une clé qui équivaut à un fichier ou une clé qui n'existe pas. J'ai essayé l'approche préférée ci-dessus et j'ai échoué sur les clés parent.

J'ai remarqué que juste pour attraper l'exception en utilisant, botocore.exceptions.ClientErrornous devons installer botocore. botocore occupe 36 Mo d'espace disque. Ceci est particulièrement impactant si nous utilisons les fonctions aws lambda. À la place de cela, si nous utilisons simplement l'exception, nous pouvons ignorer l'utilisation de la bibliothèque supplémentaire!

• Je valide l'extension de fichier pour être «.csv»
• Cela ne lèvera pas d'exception si le compartiment n'existe pas!
• Cela ne lèvera pas d'exception si le compartiment existe mais que l'objet n'existe pas!
• Cela lève une exception si le seau est vide!
• Cela lève une exception si le compartiment n'a pas d'autorisations!

Le code ressemble à ceci. Veuillez partager vos pensées:

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)

En suivant simplement le fil de discussion, quelqu'un peut-il conclure quel est le moyen le plus efficace de vérifier si un objet existe dans S3?

Je pense que head_object pourrait gagner car il vérifie simplement les métadonnées qui sont plus légères que l'objet lui-même

De https://www.peterbe.com/plog/fastest-way-to-find-out-if-a-file-exists-in-s3, il est indiqué que c'est la méthode la plus rapide:

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

Check-out

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

Vérifiez si une clé particulière existe dans le compartiment. Cette méthode utilise une requête HEAD pour vérifier l'existence de la clé. Renvoie: Une instance d'un objet Key ou None

de Boto S3 Docs

Vous pouvez simplement appeler bucket.get_key (keyname) et vérifier si l'objet renvoyé est None.