def get_template_parameters()

in source/lambdas/pipeline_orchestration/lambda_helpers.py [0:0]
60 lines of code
10 McCabe index (conditional complexity)

def get_template_parameters(event: Dict[str, Any], is_multi_account: bool, stage: str = None) -> List[Tuple[str, str]]:
    pipeline_type = event.get("pipeline_type")
    region = os.environ["REGION"]

    kms_key_arn = get_stage_param(event, "kms_key_arn", stage)
    common_params = [
        ("AssetsBucket", os.environ["ASSETS_BUCKET"]),
        ("KmsKeyArn", kms_key_arn),
        ("BlueprintBucket", os.environ["BLUEPRINT_BUCKET"]),
    ]

    # realtime params
    realtime_params = (
        [
            *common_params,
            *get_common_realtime_batch_params(event, region, stage),
            *get_realtime_specific_params(event, stage),
        ]
        if pipeline_type in ["byom_realtime_builtin", "byom_realtime_custom"]
        else None
    )
    # batch params
    batch_params = (
        [
            *common_params,
            *get_common_realtime_batch_params(event, region, stage),
            *get_batch_specific_params(event, stage),
        ]
        if pipeline_type in ["byom_batch_builtin", "byom_batch_custom"]
        else None
    )

    # create pipeline_type -> parameters map
    pipelines_params = {
        "byom_realtime_builtin": realtime_params,
        "byom_realtime_custom": realtime_params,
        "byom_batch_builtin": batch_params,
        "byom_batch_custom": batch_params,
        "byom_data_quality_monitor": [*common_params, *get_model_monitor_params(event, region, stage)]
        if pipeline_type == "byom_data_quality_monitor"
        else None,
        "byom_model_quality_monitor": [
            *common_params,
            *get_model_monitor_params(event, region, stage, monitoring_type="ModelQuality"),
        ]
        if pipeline_type == "byom_model_quality_monitor"
        else None,
        "byom_model_bias_monitor": [
            *common_params,
            *get_model_monitor_params(event, region, stage, monitoring_type="ModelBias"),
        ]
        if pipeline_type == "byom_model_bias_monitor"
        else None,
        "byom_model_explainability_monitor": [
            *common_params,
            *get_model_monitor_params(event, region, stage, monitoring_type="ModelExplainability"),
        ]
        if pipeline_type == "byom_model_explainability_monitor"
        else None,
        "byom_image_builder": [*get_image_builder_params(event)] if pipeline_type == "byom_image_builder" else None,
    }

    # get the pipeline's paramaters
    pipeline_params = pipelines_params.get(pipeline_type)

    # return the params if not NOne, otherwise throw a BadRequest exception
    if pipeline_params:
        return pipeline_params
    else:
        raise BadRequest("Bad request format. Please provide a supported pipeline")