def sagemaker_policy_statement()

in source/lib/blueprints/byom/pipeline_definitions/iam_policies.py [0:0]

• 31 lines of code
• 7 McCabe index (conditional complexity)

def sagemaker_policy_statement(is_realtime_pipeline, endpoint_name, endpoint_name_provided):
    actions = [
        "sagemaker:CreateModel",
        "sagemaker:DescribeModel",  # NOSONAR: permission needs to be repeated for clarity
        "sagemaker:DeleteModel",
    ]
    resources = [f"{sagemaker_arn_prefix}:model/mlopssagemakermodel*"]

    if is_realtime_pipeline:
        # extend actions
        actions.extend(
            [
                "sagemaker:CreateEndpointConfig",
                "sagemaker:DescribeEndpointConfig",  # NOSONAR: permission needs to be repeated for clarity
                "sagemaker:DeleteEndpointConfig",
                "sagemaker:CreateEndpoint",
                "sagemaker:DescribeEndpoint",  # NOSONAR: permission needs to be repeated for clarity
                "sagemaker:DeleteEndpoint",
            ]
        )

        # if a custom endpoint_name is provided, use it. Otherwise, use the generated name
        endpoint = core.Fn.condition_if(
            endpoint_name_provided.logical_id, endpoint_name.value_as_string, "mlopssagemakerendpoint*"
        ).to_string()

        # extend resources and add
        resources.extend(
            [
                f"{sagemaker_arn_prefix}:endpoint-config/mlopssagemakerendpointconfig*",
                f"{sagemaker_arn_prefix}:endpoint/{endpoint}",
            ]
        )
    return iam.PolicyStatement(
        actions=actions,
        resources=resources,
    )