def lambda_handler()

in source/lib/blueprints/byom/lambdas/create_update_cf_stackset/main.py [0:0]

• 39 lines of code
• 6 McCabe index (conditional complexity)

def lambda_handler(event, context):
    """The Lambda function handler

    If a continuing job then checks the CloudFormation stackset and its instances status
    and updates the job accordingly.

    If a new job then kick of an update or creation of the target
    CloudFormation stackset and its instances.

    Args:
        event: The event passed by Lambda
        context: The context passed by Lambda

    """
    job_id = None
    try:
        # Extract the Job ID
        job_id = event["CodePipeline.job"]["id"]

        # Extract the Job Data
        job_data = event["CodePipeline.job"]["data"]

        # Get user paramameters
        # User data is expected to be passed to lambda , for example:
        # {"stackset_name": "model2", "artifact":"SourceArtifact",
        # "template_file":"realtime-inference-pipeline.yaml",
        # "stage_params_file":"staging-config.json",
        # "account_ids":["<account_id>"], "org_ids":["<org_unit_id>"],
        # "regions":["us-east-1"]}
        params = get_user_params(job_data)

        # Get the list of artifacts passed to the function
        artifacts = job_data["inputArtifacts"]
        # Extract parameters
        stackset_name = params["stackset_name"]
        artifact = params["artifact"]
        template_file = params["template_file"]
        stage_params_file = params["stage_params_file"]
        account_ids = params["account_ids"]
        org_ids = params["org_ids"]
        regions = params["regions"]

        if "continuationToken" in job_data:
            logger.info(f"Checking the status of {stackset_name}")
            # If we're continuing then the create/update has already been triggered
            # we just need to check if it has finished.
            check_stackset_update_status(job_id, stackset_name, account_ids[0], regions[0], cf_client, cp_client)

        else:
            logger.info(f"Creating StackSet {stackset_name} and its instances")
            # Get the artifact details
            artifact_data = find_artifact(artifacts, artifact)
            # Get S3 client to access artifact with
            s3 = setup_s3_client(job_data)
            # Get the JSON template file out of the artifact
            template, stage_params = get_template(s3, artifact_data, template_file, stage_params_file)
            logger.info(stage_params)
            # Kick off a stackset update or create
            start_stackset_update_or_create(
                job_id,
                stackset_name,
                template,
                json.loads(stage_params),
                account_ids,
                org_ids,
                regions,
                cf_client,
                cp_client,
            )

    except Exception as e:
        logger.error(f"Error in create_update_cf_stackset lambda functions: {str(e)}")
        traceback.print_exc()
        put_job_failure(job_id, "Function exception", cp_client)
        raise e