def _copy_datazone_subscriptions()

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

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

def _copy_datazone_subscriptions(domain_id, environment_id, datazone, byor_role, execute_flag):
    """
    Copy Subscription Targets and Subscription Grants to the new BYOR Role
    
    Steps:
        1. List all subscription targets for the environment
        2. For each subscription target, list all subscription grants
        3. Delete each subscription grant
        4. Update the subscription target with the BYOR Role as the authorized principal
        5. Create new subscription grants for the new subscription target
    """
    print(f"Checking and copying subscription targets and grants for environment `{environment_id}`...\n")
    sub_target_paginator = datazone.get_paginator('list_subscription_targets')
    for sub_target_page in sub_target_paginator.paginate(domainIdentifier=domain_id, environmentIdentifier=environment_id):
        for subscription_target in sub_target_page['items']:
            target_id = subscription_target['id']
            print(f"Checking and copying subscription grants for subscription target `{target_id}`...\n")
            sub_grants_list = []
            sub_grant_paginator = datazone.get_paginator('list_subscription_grants')
            for sub_grant_page in sub_grant_paginator.paginate(domainIdentifier=domain_id, subscriptionTargetId=target_id):
                for subscription_grant in sub_grant_page['items']:
                    sub_grants_list.append(subscription_grant)
            print(f"List all Subscription grants for subscription target `{target_id}`:")
            pprint(sub_grants_list)

            # Delete all subscription grants
            for sub_grant in sub_grants_list:
                if execute_flag:
                    print(f"Calling delete subscription grant {sub_grant['id']} API... \n")
                    datazone.delete_subscription_grant(
                        domainIdentifier=domain_id,
                        identifier=sub_grant['id']
                    )
                    wait_for_subscription_grant_deletion(
                        datazone=datazone,
                        domain_id=domain_id,
                        grant_id=sub_grant['id']
                    )
                    print(f"Deleted subscription grant {sub_grant['id']} successfully \n")

            # Update subscription target with the BYOR Role
            if execute_flag:
                # In rare case after deleting all subscription grants, we still get rejected to update subscription target.
                # Add wait time bellow for safe.
                time.sleep(10)
                datazone.update_subscription_target(
                    domainIdentifier=domain_id,
                    environmentIdentifier=environment_id,
                    identifier=subscription_target['id'],
                    authorizedPrincipals=[byor_role['Role']['Arn']]
                )
            
                # Create all subscription grants which were deleted earlier
                for sub_grant in sub_grants_list:
                    create_response = datazone.create_subscription_grant(
                        domainIdentifier=domain_id,
                        environmentIdentifier=environment_id,
                        subscriptionTargetIdentifier=target_id,
                        grantedEntity={
                            'listing': {
                                'identifier': sub_grant['grantedEntity']['listing']['id'],
                                'revision': sub_grant['grantedEntity']['listing']['revision'],
                            }
                        }
                    )
                    print(f"Created new subscription grants successfully: {create_response} \n")