Tools.py

Go to the documentation of this file.

import os, json
import boto3
import requests
from datetime import date, datetime
from dotenv import load_dotenv
 
load_dotenv()
 
 
ACCESS_ID, ACCESS_KEY = os.environ.get("AWS_KEY_ID"), os.environ.get("AWS_KEY_SECRET")
 
 
s3_client = boto3.client("s3", aws_access_key_id=ACCESS_ID, aws_secret_access_key= ACCESS_KEY)
 
def gen_file_list_drive(final_path):
    """Receives the final path where files were uploaded and make a list of them.
 
    Args:
        final_path (str): s3 path where files were uploaded
    Returns: files (list)
    """
    files = []
    bucket = final_path.split("/",1)[0]
    path = final_path.split("/",1)[1]
    obj = s3_client.list_objects(Bucket = bucket, Prefix=path)
    for file in obj["Contents"]:
        if file["Key"].endswith("/"):
            pass
        else:
            full_path = bucket+"/"+file["Key"]
            files.append(full_path)
    return files
 
def gen_file_list_event(event, final_path):
    """Checks which files in the input were uploaded successful.
 
    Args:
        event (dict): is a dictionary with all client and files information
        final_path (str): s3 path where files were uploaded
    Returns: files_out (list)
    """
    files_out = []
    files = event["file"]
    if type(event["file"]) == str:
        files = [event["file"]]
    for file in files:
        full_path = final_path+file
        files_out.append(full_path)
    return files_out
 
def send_notification(module, webhook_url, status, files=None, error=None):
    """Sends a success/fail notification to some webhook.
 
    Args:
        module (str): current module
        webhook_url (str): webhook url 
        status (str): final status of execution
        files (list): files uploaded to s3. None when status =! successful
        error (str): error show while procedure fails. None when status = successful
    Returns: Nothing
    """
    now = datetime.now()
    date = now.strftime("%Y-%m-%dT%H:%M:%SZ")
    if status == "Successful":
        message = {
            "date": date,
            "module": module,
            "status": "Finished",
            "message": "Files uploader has finished. Reports are on S3.", 
            "files": files
        }
    else:
        message = {
            "date": date,
            "module": module,
            "status": "Failed",
            "message": "Files could not be uploaded. Copy the error bellow and contact support to fix it.", 
            "error": error
        }
    req = requests.post(webhook_url, data=json.dumps(message, indent=4), headers={"Content-type": "application/json"})
    print("Request sent to OAS Webhook")