from - python aws sdk

comprobar si existe una clave en un cubo en s3 usando boto3

Question

Me gustaría saber si existe una clave en boto3. Puedo recorrer el contenido del cubo y verificar la clave si coincide.

Pero eso parece más largo y una exageración. Los documentos oficiales de Boto3 indican explícitamente cómo hacer esto.

Puede ser que me falta lo obvio. ¿Alguien puede señalarme cómo puedo lograr esto?

Answer 1

Aquí hay una solución que me funciona. Una advertencia es que sé el formato exacto de la clave con anticipación, por lo que solo enumero el archivo único

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

Answer 2

El objeto boto.s3.key.Key Boto 2 solía tener un método exists que verificaba si la clave existía en S3 haciendo una solicitud HEAD y mirando el resultado, pero parece que ya no existe. Tienes que hacerlo tú mismo:

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() realiza una solicitud HEAD para una sola clave, que es rápida, incluso si el objeto en cuestión es grande o tiene muchos objetos en su depósito.

Por supuesto, es posible que esté verificando si el objeto existe porque planea usarlo. Si ese es el caso, puede olvidarse de la load() y hacer un get() o download_file() directamente, luego maneje el caso de error allí.

Answer 3

En Boto3, si está buscando una carpeta (prefijo) o un archivo usando list_objects. Puede usar la existencia de ''Contenido'' en el dict de respuesta como una verificación de si el objeto existe. Es otra forma de evitar el intento / excepto las capturas como sugiere @EvilPuppetMaster

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

Answer 4

FWIW, aquí están las funciones muy simples que estoy usando

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

Answer 5

Hay una manera simple por la cual podemos verificar si el archivo existe o no en el depósito S3. No necesitamos usar una excepción para esto

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

Answer 6

La forma más fácil que encontré (y probablemente la más eficiente) es esta:

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

Answer 7

No solo client sino también 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'')

Answer 8

No soy un gran fanático del uso de excepciones para el flujo de control. Este es un enfoque alternativo que funciona en 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")

Answer 9

Noté que solo para detectar la excepción usando botocore.exceptions.ClientError necesitamos instalar botocore. botocore ocupa 36 millones de espacio en disco. Esto es particularmente impactante si usamos funciones aws lambda. En lugar de eso, si solo usamos la excepción, ¡podemos omitir el uso de la biblioteca adicional!

• Estoy validando que la extensión del archivo sea ''.csv''
• ¡Esto no arrojará una excepción si el cubo no existe!
• ¡Esto no arrojará una excepción si el depósito existe pero el objeto no existe!
• ¡Esto arroja una excepción si el cubo está vacío!
• ¡Esto arroja una excepción si el cubo no tiene permisos!

El código se ve así. Por favor comparte tus pensamientos:

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)

Answer 10

Para boto3, ObjectSummary se puede usar para verificar si existe un objeto.

Contiene el resumen de un objeto almacenado en un bucket de Amazon S3. Este objeto no contiene los metadatos completos del objeto ni ninguno de sus contenidos.

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

En ObjectSummary.load

Llama a s3.Client.head_object para actualizar los atributos del recurso ObjectSummary.

Esto muestra que puede usar ObjectSummary lugar de Object si planea no usar get() . La función load() no recupera el objeto, solo obtiene el resumen.

Answer 11

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

Answer 12

Puede usar S3Fs , que es esencialmente un contenedor alrededor de boto3 que expone las operaciones típicas de estilo de sistema de archivos:

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

Answer 13

Revisa

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

Verifique si existe una clave en particular dentro del cubo. Este método utiliza una solicitud HEAD para verificar la existencia de la clave. Devuelve: una instancia de un objeto clave o ninguno

de Boto S3 Docs

Simplemente puede llamar a bucket.get_key (keyname) y verificar si el objeto devuelto es None.

Answer 14

Si busca una clave que sea equivalente a un directorio, es posible que desee este enfoque

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

Esto funciona para una clave principal o una clave que equivale a un archivo o una clave que no existe. Probé el enfoque favorito anterior y fallé en las claves principales.

Answer 15

Si tiene menos de 1000 en un directorio o depósito, puede obtener un conjunto de ellos y luego verificar si dicha clave en este conjunto:

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

Dicho código funciona incluso si my/dir no existe.

http://boto3.readthedocs.io/en/latest/reference/services/s3.html#S3.Client.list_objects_v2

Answer 16

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)

Answer 17

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