def process_cloudformation_event()

in aws_sra_examples/solutions/s3/s3_block_account_public_access/lambda/src/app.py [0:0]

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

def process_cloudformation_event(event: Dict[str, Any], context: Any) -> str:
    """Process Event from AWS CloudFormation.

    Args:
        event: event data
        context: runtime information

    Returns:
        AWS CloudFormation physical resource id
    """
    params = get_validated_parameters(event)

    management_account_session = boto3.session.Session()
    set_configuration_ssm_parameters(management_account_session, params)

    enable_block_public_acls = (params.get("ENABLE_BLOCK_PUBLIC_ACLS", "true")).lower() in "true"
    enable_ignore_public_acls = (params.get("ENABLE_IGNORE_PUBLIC_ACLS", "true")).lower() in "true"
    enable_block_public_policy = (params.get("ENABLE_BLOCK_PUBLIC_POLICY", "true")).lower() in "true"
    enable_restrict_public_buckets = (params.get("ENABLE_RESTRICT_PUBLIC_BUCKETS", "true")).lower() in "true"

    if params["action"] in ("Add"):
        account_ids = get_all_organization_accounts()

        thread_cnt = MAX_THREADS
        if MAX_THREADS > len(account_ids):
            thread_cnt = max(len(account_ids) - 2, 1)

        processes = []
        with ThreadPoolExecutor(max_workers=thread_cnt) as executor:
            for account_id in account_ids:
                processes.append(
                    executor.submit(
                        process_put_account_public_access_block,
                        management_account_session,
                        params,
                        account_id,
                        enable_block_public_acls,
                        enable_ignore_public_acls,
                        enable_block_public_policy,
                        enable_restrict_public_buckets,
                    )
                )
            for future in as_completed(processes, timeout=60):
                try:
                    future.result()
                except Exception as error:
                    LOGGER.error(f"{error}")
                    raise ValueError(f"There was an error updating the S3 account public access settings")
    else:
        ssm_client: SSMClient = management_account_session.client("ssm")
        delete_ssm_parameter(ssm_client, SSM_PARAMETER_PREFIX)

    return (
        f"S3PublicAccessBlock-{params['ENABLE_BLOCK_PUBLIC_ACLS']}"
        f"-{params['ENABLE_IGNORE_PUBLIC_ACLS']}"
        f"-{params['ENABLE_BLOCK_PUBLIC_POLICY']}"
        f"-{params['ENABLE_RESTRICT_PUBLIC_BUCKETS']}"
    )