glider_upload.py

Go to the documentation of this file.

import sys
import pymongo
import boto3
from dotenv import load_dotenv
import os
import logging
import json
import time
 
path = os.path.dirname(sys.argv[0])
load_dotenv('{}/.env'.format(path)) #full_path
 
file = sys.argv[1]
file_id = sys.argv[2]
client_id = sys.argv[3]
exchange_rates = sys.argv[4]
print(file)
# filename = sys.argv[5]
myclient = pymongo.MongoClient(os.environ.get('MONGO_CONNECTION'))
mydb = myclient["myFirstDatabase"]
mycol = mydb["files"]
 
def addToMongo(s3_path, file_id, client_id, filepath):
    file_dict = { "file_db_id": file_id, "status": "pending", "client_id":  client_id, 'name': os.path.basename(filepath), "s3_path":s3_path}
    logging.info('inserting_file')
    result = mycol.replace_one({"file_db_id": file_id}, file_dict, upsert=True)
 
def upload_to_s3(file, client_id):
    s3_client = boto3.client('s3')
    object_name = '{}/{}'.format(client_id,os.path.basename(file))
    try:
        response = s3_client.upload_file(file, os.environ.get('BUCKET_INPUT'), object_name)
    except Exception as e:
        logging.error(e)
        return False
    return object_name
 
def trigger_step_function_ingest(file_id, client_id, file):
    step_functions_client = boto3.client("stepfunctions")
    input_parameters = {
        "client_id": client_id,
        "bucket":[os.environ.get('BUCKET_INPUT'), os.environ.get('BUCKET_OUTPUT')],
        "path":["{}".format(client_id), "{}".format(client_id)],
        "currency": {"EUR": 1, "USD": 1, "GBP": 1},
        "files": [{"file_id": file_id, "file": file}]
    }
 
    execution_name = '{}-{}-{}-{}'.format(file_id,client_id,'ottito', int(time.time()))
    response = step_functions_client.start_execution(
        stateMachineArn=os.environ.get('INGESTION_STEP_FUNCTION_ARN'),
        name=execution_name,
        input=json.dumps(input_parameters),
    )
    result = mycol.update_one({"file_db_id": file_id},{"$set": {"status": "processing"}})
 
# s3_path = upload_to_s3(file, client_id)
# addToMongo(s3_path, file_id, client_id, file)
# trigger_step_function_ingest(file_id, client_id, os.path.basename(file))