def wait_for_view()

in notebooks/Utilities/finspace.py [0:0]

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

    def wait_for_view(self, dataset_id: str, view_id: str, sleep_sec=10):
        """
        function that will continuously poll the view creation to ensure it completes or fails before returning.

        :param dataset_id: GUID of the dataset
        :type: str

        :param view_id: GUID of the view
        :type: str

        :param sleep_sec: seconds to wait between checks
        :type: int

        """
        while True:
            list_views_resp = self.client.list_materialization_snapshots(datasetId=dataset_id, maxResults=100)
            matched_views = list(filter(lambda d: d['id'] == view_id, list_views_resp['materializationSnapshots']))

            if len(matched_views) != 1:
                size = len(matched_views)
                raise Exception(f"Unexpected error: found {size} views that match the view Id: {view_id}")

            status = matched_views[0]['status']
            if status == 'SUCCESS':
                print(f"View complete")
                break
            elif status == 'PENDING' or status == 'RUNNING':
                print(f"View status is still PENDING, continue to wait till finish...")
                time.sleep(sleep_sec)
                continue
            else:
                raise Exception(f"Bad view status: {status}, failing now.")