MVAD Polling

mvad_polling.py

class MVADPolling(LongRunningOperation):
    """Implements MVAD train polling."""

    def __init__(self):
        self._model_id = None

    def can_poll(self, pipeline_response: "PipelineResponseType") -> bool:
        # This method is used for algorithm chaining, in MVAD we don't need
        # it since this is the only one we will register
        return True

    def get_polling_url(self) -> str:
        return f"/multivariate/models/{self._model_id}"

    def set_initial_status(self, pipeline_response: "PipelineResponseType") -> str:
        """Process first response after initiating long running operation.
        :param azure.core.pipeline.PipelineResponse response: initial REST call response.
        """
        response = pipeline_response.http_response
        self._model_id = response.json()["modelId"]

        # The first call is never final, set to poll
        return "InProgress"

    def get_status(self, pipeline_response: "PipelineResponseType") -> str:
        response = pipeline_response.http_response
        status = response.jon()["modelInfo"]["status"]
        if status == "READY":
            # Re-map this one, so we don't have to override the loop system
            return "succeeded"
        return status  # Maybe FAILED or something else
      
    def get_final_get_url(self, pipeline_response: "PipelineResponseType") -> Optional[str]:
        # MVAD do not need a final GET call
        return None