UploadReports.py

Go to the documentation of this file.

import os
import boto3
import pyshorteners
from dotenv import load_dotenv
from botocore.exceptions import ClientError
load_dotenv()
 
ACCESS_ID = os.environ.get("AWS_KEY_ID")
ACCESS_KEY = os.environ.get("AWS_KEY_SECRET")
REGION = os.environ.get("REGION")  
ACCESS_TOKEN = os.environ.get("BITLY_TOKEN")  
s3_client = boto3.client("s3", aws_access_key_id=ACCESS_ID, aws_secret_access_key= ACCESS_KEY, region_name=REGION)
 
def create_presigned_url(bucket, object, expiration=86400):#86400
    """Generate a presigned URL to share an S3 object via lambda invoke
 
    Args:
        bucket (str): bucket where report was uploaded
        object (str): full report object name 
        expiration (int): time in seconds for the presigned URL to remain valid
    Returns: response (str): If error, returns None.
    """
    params={'Bucket': bucket, 'Key': object}
    try:
        response = s3_client.generate_presigned_url('get_object', Params=params, ExpiresIn=expiration)
    except ClientError as e:
        return None
    return response
    # get_log = True 
    # function_params = { 
    #     "bucket": bucket, 
    #     "object": object, 
    #     "expiration": expiration 
    #     }    
    # client = boto3.client('lambda', aws_access_key_id=ACCESS_ID, aws_secret_access_key=ACCESS_KEY, region_name="eu-west-2") 
    # lambda_response = client.invoke( 
    #             FunctionName=function_name, 
    #             Payload=json.dumps(function_params), 
    #             LogType='Tail' if get_log else 'None')       
    # link_response = lambda_response['Payload'].read() 
    # link_response = link_response.strip()
    # link_response = json.loads(link_response)
    # response = link_response["response"]
    # return response
 
def shorten_url(long_url):
    """
    Shorts URL using TynyURL service
    
    Args:
        long_url (str): presigned url for report
    Returns: short_url.
    """
    # bitly_access_token = ACCESS_TOKEN
 
    try:
        # shortener = pyshorteners.Shortener(api_key=bitly_access_token)
        shortener = pyshorteners.Shortener()
        short_url = shortener.tinyurl.short(long_url)
        return short_url
    except Exception as e:
        print(f"Error shorting URL: {e} Returning long url") 
        return long_url
 
def upload_files_s3(bucket, s3_path, local_path, file):
    """
    Uploads all files to s3 path defined previously
 
    Args:
        local_path (str): Root path where is saved files
        files (str): File to upload
    Returns: url, s3 path.
    """
    full_path = local_path+file
    s3_full_path = s3_path+file
    s3_client.upload_file(full_path, bucket, s3_full_path)
    long_url = create_presigned_url(bucket, s3_path+file)
    url = shorten_url(long_url)
    return url, s3_full_path