En la presente entrada, AWS S3 y Luigi, módulo S3: eliminación y copiado de carpetas, realizaré la descripción de la tareas para realizar con Luigi las operaciones de eliminación y copiado de carpetas de un bucket en  Amazon S3 con Luigi. Para realizar las operaciones sobre S3 de Amazon con Luigi, necesitamos instanciar un cliente S3Client dentro de una tarea.

En nuestro caso, definiré una jerarquía de clases en donde la clase padre define la referencia a un cliente S3 de Luigi: de esta forma, las clases hijos simplemente deben de pasar los parámetros con los datos necesarios. La importación del cliente es la siguiente:

from luigi.contrib.s3 import S3Client

La cabecera de la clase con la definición de la referencia al cliente es la siguiente:

class AbstractS3Client(luigi.Task):
  """
  La clase AbstractS3Client realiza la operaciones básicas en S3.
  """
  s3config = S3Configuration()
  s3Client = S3Client(profile_name=s3config.profileName, host=s3config.awsHostS3)
  [...]

Operación de eliminación

La operación de eliminación del contenido de una carpeta de S3 se realiza empleando la función remove del cliente S3Client. Además del path de la carpeta que se quiere eliminar, es necesario el nombre del Bucket de S3.

El snippet con el código con la funcionalidad de la eliminación de la carpeta es la siguiente:

def remove(self, bucketName, path):
  """
  El método removeProcess realiza la eliminación del contenido definido en un path pasado por parámetro de un
  bucket determinado.
  :param bucketName: Nombre del bucket
  :param path: Path
  :return:
  """
  path = "s3://{bucketName}/{path}/".format(bucketName=bucketName, path=path)
  if self.s3Client.exists(path):
    logging.info("\n[AbstractS3Client.removeProcess] Verificado el path={path}".format(path=path))
    resultRemove = self.s3Client.remove(path)
    if resultRemove:
      logging.info("[AbstractS3Client.removeProcess] Eliminación de {path}: OK".format(path=path))
    else:
      logging.info("[AbstractS3Client.removeProcess] Eliminación de {path}: KO".format(path=path))
  else:
    logging.info("[AbstractS3Client.removeProcess] NO Verificado el path={path}".format(path=path))

En el snippet anterior, código que se integra en la clase AbstractS3Client, define el path de la carpeta del bucket a eliminar con la variable path; se realiza la comprobación de existencia del bucket que se quiere eliminar con la función exists; si la comprobación anterior es correcta, se realiza la eliminación con la función remove, en otro caso, se muestra un mensaje informativo.

Operación de copia

La operación de copiado del contenido de una carpeta origen a otra carpeta destino, se realiza con la función copy del cliente S3Client. Además, del path de la carpeta origen y destino, es necesario el nombre del bucket de S3.

El snippet con el código con la funcionalidad de la operación de copiado de las carpetas, es el siguiente:

def copy(self, bucketName, sourceFolder, targetFolder):
  """
  El método copyProcess realiza el copiado del contenido de una carpeta origen(sourceFolder) a una carpeta
  destino (targetFolder) de un bucket determinado (bucketName).
  :param bucketName: Nombre del bucket
  :param sourceFolder: Carpeta origen
  :param targetFolder: Carpeta destino.
  :return:
  """
  sourceFile = "s3://{bucketName}/{sourceFolder}/".format(bucketName=bucketName, sourceFolder=sourceFolder)
  if self.s3Client.isdir(sourceFile):
    targetFile = "s3://{bucketName}/{targetFolder}/".format(bucketName=bucketName, targetFolder=targetFolder)
    if self.s3Client.isdir(targetFile):
      resultadoCopiado = self.s3Client.copy(sourceFile, targetFile)
      logging.info("[AbstractS3Client.copy] Copiado contenido DESDE:{sourceFile} A:{targetFile}".format(sourceFile=sourceFile, targetFile=targetFile))
      logging.info("[AbstractS3Client.copy] Numero de ficheros copiados={resultadoCopiado0}".format(resultadoCopiado0=resultadoCopiado[0]))
      logging.info("[AbstractS3Client.copy] Tamaño copiado={resultadoCopiado1}".format(resultadoCopiado1=resultadoCopiado[1]))
    else:
      logging.info("[AbstractS3Client.copy] {targetFile} no es un directorio.".format(targetFile=targetFile))
  else:
    logging.info("[AbstractS3Client.copy] {sourceFile} no es un directorio.".format(sourceFile=sourceFile))

En el snippet anterior, código que se integra en la clase AbstractS3Client, define el path de la carpeta origen de copiado definido en la variabla sourceFile; se realiza la comprobación de existencia del directorio con la función isDir; si el path del directorio es correcto, se calcula el path de la carpeta destino de copiado definido en la variable targetFile; si el path del directorio destino es correcto empleando la función isDir, se realiza el copiado del contenido de la carpeta origen en la destino; y, de todas las comprobaciones, se muestra el mensaje de error si no se cumple.

Instancia de clases

Para ejecutar la funcionalidad descrita en el apartado anterior, es necesario definir una clase que herede de la clase AbstractS3Client. Al ser esta clase una tarea de Luigi, es necesario la implementación del método run. Si se desea que la tarea esté enlazada con otras tareas, es necesario definir el método output.

El snippet de una clase que hereda de la clase AbstractS3Client para realizar la eliminación de una carpeta es la siguiente:

class DeleteS3(AbstractS3Client):
  def run(self):
    s3Configuration = S3Configuration()
    self.remove(s3Configuration.bucketName, s3Configuration.sourceFolder)

El snippet anterior, define la clase S3Configuration la cual define las siguientes propiedades: bucketName, nombre del bucket; sourceFolder, nombre de la carpeta a eliminar; y, además, define la invocación del método remove para ejecutar la eliminación.

El snippet de una clase que hereda de la clase AbstractS3Client para realizar el copiado del contenido de una carpeta es el siguiente:

class CopyS3(AbstractS3Client):
  def run(self):
    s3Configuration = S3Configuration()
    self.copyProcess(s3Configuration.bucketName, s3Configuration.sourceFolder, s3Configuration.targetFolder)

El snippet anterior, define la clase S3Configuration la cual define las siguientes propiedades: bucketName, nombre del bucket; sourceFolder, nombre de la carpeta origen a copiar; targetFolder, carpeta destino de copiado; y, además, define la invocación del método remove para ejecutar la eliminación.

Las operaciones con Amazon S3 son sencillas de realizar con Luigi. Lo que se tiene que tener claro son todos aquellos datos de configuración para la ejecución de las operaciones.

Para el lector interesado, las entrada publicadas hasta la fecha sobre Luigi son las siguientes:

• Luigi