trigger.py

Go to the documentation of this file.

import os
import boto3
import json
import time
from dotenv import load_dotenv
load_dotenv()
 
def trigger_step_function_ingest(status, client_id, results_bucket, results_path):
    ACCESS_ID = os.environ.get("AWS_KEY_ID")
    ACCESS_KEY = os.environ.get("AWS_KEY_SECRET")
    REGION = os.environ.get("REGION")
    step_functions_client = boto3.client("stepfunctions", aws_access_key_id=ACCESS_ID, aws_secret_access_key=ACCESS_KEY, region_name=REGION)
    input_parameters = {
        "status": status,
        "client_id": client_id,
        "results_bucket": results_bucket,
        "results_path": results_path
    }
    client_id = client_id.replace(" ", "-")
    path = results_path.replace("/", "-")
    execution_name = '{}-{}-{}-{}'.format(path,client_id,'ottito', int(time.time()))
    response = step_functions_client.start_execution(
        stateMachineArn=os.environ.get('RESULTS_STEP_FUNCTION_ARN'),
        name=execution_name,
        input=json.dumps(input_parameters)
    )
 
def lambda_handler(event, context):
    status = event["status"]
    client_id = event["client_id"]
    # formats = event["format"]
    results_bucket = event["results_bucket"]
    results_path = event["results_path"]
    trigger_step_function_ingest(status, client_id, results_bucket, results_path)
    return "Catalog in progress"