Crud Python AWS S3

Meu objetivo nesse post é difundir as tecnologias Python, AWS e S3. Será desenvolvido um crud no bucket do S3 afim de realizar operações upload, download, put, read e delete de um arquivo no S3 no AWS.

Será necessário configurar S3 no AWS e logo após criar os arquivos em Python para realizar a operação.

O projeto esta armazenado no github: https://github.com/alexjosesilva/crud-python-aws-s3

1. Configurar o AWS S3

Criar usuário e política de segurança

Criar upload_file.py

import boto3

#o arquivo configure esta as chaves de acesso para o bucke 
#aws_access_key_id e aws_secret_access_key

from configure import *
class UploadFile(object):
    
    def __init__(self) -> None:
        pass
    
    def upload_file(self):
    
        bucket_name = "bucketuse2023"
        file_name = "input/business-financial-data-september-2022-quarter.csv"
        object_name = 'input/business-financial-data-september-2022-quarter.csv'
        
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            s3.Bucket(bucket_name).upload_file(object_name, file_name)
        except Exception as err:
            print("An exception occurred: upload: ", err)

Criar get_file.py

import boto3

#o arquivo configure esta as chaves de acesso para o bucke 
#aws_access_key_id e aws_secret_access_key
from configure import *

class GetFile(object):
    
    def __init__(self) -> None:
        pass
    
    def get_file(self):
        bucket_name = "bucketuse2023"
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            s3.Bucket(bucket_name).download_file('input/s2.csv', 'input/s2.csv')
        except Exception as err:
            print("An exception occurred: upload: ", err)

Criar o list_file.py

import boto3
from configure import *

class ReadFile(object):
    
    def __init__(self) -> None:
        pass

    def read_file(self):
        file_name = 'input/s2.csv'
        bucket_name = "bucketuse2023"
    
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            obj = s3.Object(bucket_name,file_name)
            data=obj.get()['Body'].read()
            print(data)
            
        except Exception as err:
            print("An exception occurred: readfile: ", err)

Criar delete_file.py

import boto3
from configure import *

file_name = 'input/s2.csv'
bucket_name = "bucketuse2023"

try:
    s3 = boto3.resource('s3',
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY
    )
    s3.Object(bucket_name, file_name).delete()
    
except Exception as err:
    print("An exception occurred: upload: ", err)

Criar leitura de arquivos nas pastas ou por tipo

import boto3
import re 
from configure import *

class ReadFilesInBucket(object):
    
    def __init__(self) -> None:
        pass
    #buscar todos os arquivos
    def list_all_files_in_bucket(self):
        bucket_name = "bucketuse2023"
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            
            my_bucket = s3.Bucket(bucket_name)
            for my_bucket_obj in my_bucket.objects.all():
                print(my_bucket_obj.key)
                
        except Exception as err:
            print("An exception occurred: read name files in bucket in aws s3: ", err)
        
    #buscar na pasta 
    def list_filtre_files_in_bucket(self, filtre):
        bucket_name = "bucketuse2023"
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            
            my_bucket = s3.Bucket(bucket_name)
            for my_bucket_obj in my_bucket.objects.filter(Prefix=filtre):
                print(my_bucket_obj.key)
                
        except Exception as err:
            print("An exception occurred: read name files in bucket in aws s3: ", err)  
    
    #buscar    
    def list_types_files_in_bucket(self, type_files):
        bucket_name = "bucketuse2023"
        try:
            s3 = boto3.resource('s3',
                aws_access_key_id=AWS_ACCESS_KEY_ID,
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY
            )
            
            my_bucket = s3.Bucket(bucket_name)
            substring =  "\d"
            for my_bucket_obj in my_bucket.objects.all():
                if re.search(substring,  my_bucket_obj.key):  
                    print(my_bucket_obj.key)
                
        except Exception as err:
            print("An exception occurred: read name files in bucket in aws s3: ", err)