def _obtain_long_running_operation()

in src/python/tensorflow_cloud/tuner/vizier_client.py [0:0]

• 32 lines of code
• 4 McCabe index (conditional complexity)

    def _obtain_long_running_operation(self, resp):
        """Obtain the long-running operation."""
        op_id = resp["name"].split("/")[-1]
        operation_name = "projects/{}/locations/{}/operations/{}".format(
            self.project_id, self.region, op_id
        )
        try:
            get_op = (
                self.service_client.projects()
                .locations()
                .operations()
                .get(name=operation_name)
            )
            operation = get_op.execute()
        except errors.HttpError as e:
            tf.get_logger().info("GetLongRunningOperations failed.")
            raise e

        polling_secs = 1
        num_attempts = 0
        while not operation.get("done"):
            sleep_time = self._polling_delay(num_attempts, polling_secs)
            num_attempts += 1
            tf.get_logger().info(
                "Waiting for operation; attempt {}; "
                "sleeping for {} seconds".format(
                    num_attempts, sleep_time
                )
            )
            time.sleep(sleep_time.total_seconds())
            if num_attempts > 30:  # about 10 minutes
                raise RuntimeError("GetLongRunningOperations timeout.")
            operation = get_op.execute()
        return operation