def _get_S3_registered_locations()

in migration/bring-your-own-gdc-assets/bring_your_own_gdc_assets.py [0:0]

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

def _get_S3_registered_locations(lf_client):
    """
    Get all registered S3 locations
    """
    registered_locations = []
    next_token = None

    try:
        while True:
            params = {}
            if next_token:
                params['NextToken'] = next_token

            response = lf_client.list_resources(**params)

            # Process resources in current page
            for resource in response.get('ResourceInfoList', []):
                resource_arn = resource.get('ResourceArn', '')
                if 's3:::' in resource_arn:
                    registered_locations.append(s3_arn_to_s3_path(resource_arn))

            # Check if there are more resources to fetch
            next_token = response.get('NextToken')
            if not next_token:
                break

    except ClientError as e:
        print(f"Error calling Lake Formation list_resources api to fetch registered S3 locations: {str(e)}")
        raise e

    return registered_locations