def _copy_lakeformation_opt_ins()

in migration/bring-your-own-role/byor.py [0:0]

• 38 lines of code
• 8 McCabe index (conditional complexity)

def _copy_lakeformation_opt_ins(lakeformation, source_role_arn, destination_role_arn, execute_flag):
    print(f"Checking and copying lakeformation opt ins associated with role `{source_role_arn}` to role `{destination_role_arn}`...\n")
    opt_in_list_to_copy = []
    response = lakeformation.list_lake_formation_opt_ins(
        Principal={
            'DataLakePrincipalIdentifier': source_role_arn
        }
    )
    for opt_in in response['LakeFormationOptInsInfoList']:
        opt_in_list_to_copy.append(opt_in)
    while response.get('NextToken'):
        response = lakeformation.list_lake_formation_opt_ins(NextToken=response['NextToken'])
        for opt_in in response['LakeFormationOptInsInfoList']:
            opt_in_list_to_copy.append(opt_in)

    for opt_in_to_copy in opt_in_list_to_copy:
        print(f"Copying LakeFormation Opt In:")
        pprint(opt_in_to_copy)
        print(f"to new role: {destination_role_arn}...\n")
        if execute_flag:
            try:
                lakeformation.create_lake_formation_opt_in(
                    Principal={
                        'DataLakePrincipalIdentifier': destination_role_arn
                    },
                    Resource=_filter_lakeformationsource(opt_in_to_copy['Resource']),
                )
            except ClientError as e:
                if e.response['Error']['Code'] == 'InvalidInputException':
                    print(f"Opt-in already exists, skipping...\n")
                else:
                    raise e
            print(f"Successfully copy LakeFormation Opt In:")
            pprint(opt_in_to_copy)
            print(f"to new role: {destination_role_arn} \n")
        else:
            print(f"Skipping copy LakeFormation Opt In:")
            pprint(opt_in_to_copy)
            print(f"to new role: {destination_role_arn}, set --execute flag to True to do the actual update.\n")