def _copy_lakeformation_grants()

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

• 35 lines of code
• 10 McCabe index (conditional complexity)

def _copy_lakeformation_grants(lakeformation, source_role_arn, destination_role_arn, execute_flag, script_option):
    print(f"Checking and copying lakeformation grants associated with role `{source_role_arn}` to role `{destination_role_arn}`...\n")
    grants_list_to_copy = []
    response = lakeformation.list_permissions()
    for grant in response['PrincipalResourcePermissions']:
        if grant['Principal']['DataLakePrincipalIdentifier'] == source_role_arn:
            grants_list_to_copy.append(grant)
    while response.get('NextToken'):
        response = lakeformation.list_permissions(NextToken=response['NextToken'])
        for grant in response['PrincipalResourcePermissions']:
            if grant['Principal']['DataLakePrincipalIdentifier'] == source_role_arn:
                grants_list_to_copy.append(grant)
    if not grants_list_to_copy:
        if script_option == ROLE_REPLACEMENT:
            # Auto generated Project role has grants associated with it in some project profiles but not all, log out warn message
            print(f"WARN: No grants found associated with role {source_role_arn}, skipping copy... Please make sure you added script executor as LakeFormation Data lake administrators properly.\n")

    for grant_to_copy in grants_list_to_copy:
        print(f"Copying LakeFormation Grant:")
        pprint(grant_to_copy)
        print(f"to new role: {destination_role_arn}...\n")
        if execute_flag:
            lakeformation.grant_permissions(
                Principal={
                    'DataLakePrincipalIdentifier': destination_role_arn
                },
                Resource=_filter_lakeformationsource(grant_to_copy['Resource']),
                Permissions=grant_to_copy['Permissions'],
                PermissionsWithGrantOption=grant_to_copy['PermissionsWithGrantOption']
            )
            print(f"Successfully copy LakeFormation Grant:")
            pprint(grant_to_copy)
            print(f"to new role: {destination_role_arn} \n")
        else:
            print(f"Skipping copy LakeFormation Grant:")
            pprint(grant_to_copy)
            print(f"to new role: {destination_role_arn}, set --execute flag to True to do the actual update.\n")