Refactor CloudApiClient (and ToolpathUploader) to use HttpRequestManager

Has the benefit of a more unified Http request management + timeouts

CURA-7290

plugins/UM3NetworkPrinting/src/Cloud/CloudApiClient.py

@@ -6,11 +6,15 @@ from time import time
 from typing import Callable, List, Type, TypeVar, Union, Optional, Tuple, Dict, Any, cast
 
 from PyQt5.QtCore import QUrl
-from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply, QNetworkAccessManager
+from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply
 
 from UM.Logger import Logger
+from UM.TaskManagement.HttpRequestManager import HttpRequestManager
+from UM.TaskManagement.HttpRequestScope import JsonDecoratorScope
 from cura.API import Account
+from cura.CuraApplication import CuraApplication
 from cura.UltimakerCloud import UltimakerCloudAuthentication
+from cura.UltimakerCloud.UltimakerCloudScope import UltimakerCloudScope
 from .ToolPathUploader import ToolPathUploader
 from ..Models.BaseModel import BaseModel
 from ..Models.Http.CloudClusterResponse import CloudClusterResponse
@@ -33,16 +37,20 @@ class CloudApiClient:
     CLUSTER_API_ROOT = "{}/connect/v1".format(ROOT_PATH)
     CURA_API_ROOT = "{}/cura/v1".format(ROOT_PATH)
 
+    DEFAULT_REQUEST_TIMEOUT = 10  # seconds
+
     # In order to avoid garbage collection we keep the callbacks in this list.
-    _anti_gc_callbacks = []  # type: List[Callable[[], None]]
+    _anti_gc_callbacks = []  # type: List[Callable[[Any], None]]
 
     ## Initializes a new cloud API client.
     #  \param account: The user's account object
     #  \param on_error: The callback to be called whenever we receive errors from the server.
-    def __init__(self, account: Account, on_error: Callable[[List[CloudError]], None]) -> None:
+    def __init__(self, app: CuraApplication, on_error: Callable[[List[CloudError]], None]) -> None:
         super().__init__()
-        self._manager = QNetworkAccessManager()
+        self._app = app
-        self._account = account
+        self._account = app.getCuraAPI().account
+        self._scope = JsonDecoratorScope(UltimakerCloudScope(app))
+        self._http = HttpRequestManager.getInstance()
         self._on_error = on_error
         self._upload = None  # type: Optional[ToolPathUploader]
 
@@ -55,16 +63,21 @@ class CloudApiClient:
     #  \param on_finished: The function to be called after the result is parsed.
     def getClusters(self, on_finished: Callable[[List[CloudClusterResponse]], Any], failed: Callable) -> None:
         url = "{}/clusters?status=active".format(self.CLUSTER_API_ROOT)
-        reply = self._manager.get(self._createEmptyRequest(url))
+        self._http.get(url,
-        self._addCallback(reply, on_finished, CloudClusterResponse, failed)
+                       scope = self._scope,
+                       callback = self._parseCallback(on_finished, CloudClusterResponse, failed),
+                       error_callback = failed,
+                       timeout = self.DEFAULT_REQUEST_TIMEOUT)
 
     ## Retrieves the status of the given cluster.
     #  \param cluster_id: The ID of the cluster.
     #  \param on_finished: The function to be called after the result is parsed.
     def getClusterStatus(self, cluster_id: str, on_finished: Callable[[CloudClusterStatus], Any]) -> None:
         url = "{}/clusters/{}/status".format(self.CLUSTER_API_ROOT, cluster_id)
-        reply = self._manager.get(self._createEmptyRequest(url))
+        self._http.get(url,
-        self._addCallback(reply, on_finished, CloudClusterStatus)
+                       scope = self._scope,
+                       callback = self._parseCallback(on_finished, CloudClusterStatus),
+                       timeout = self.DEFAULT_REQUEST_TIMEOUT)
 
     ## Requests the cloud to register the upload of a print job mesh.
     #  \param request: The request object.
@@ -72,9 +85,13 @@ class CloudApiClient:
     def requestUpload(self, request: CloudPrintJobUploadRequest,
                       on_finished: Callable[[CloudPrintJobResponse], Any]) -> None:
         url = "{}/jobs/upload".format(self.CURA_API_ROOT)
-        body = json.dumps({"data": request.toDict()})
+        data = json.dumps({"data": request.toDict()}).encode()
-        reply = self._manager.put(self._createEmptyRequest(url), body.encode())
+
-        self._addCallback(reply, on_finished, CloudPrintJobResponse)
+        self._http.put(url,
+                        scope = self._scope,
+                        data = data,
+                        callback = self._parseCallback(on_finished, CloudPrintJobResponse),
+                        timeout = self.DEFAULT_REQUEST_TIMEOUT)
 
     ## Uploads a print job tool path to the cloud.
     #  \param print_job: The object received after requesting an upload with `self.requestUpload`.
@@ -84,7 +101,7 @@ class CloudApiClient:
     #  \param on_error: A function to be called if the upload fails.
     def uploadToolPath(self, print_job: CloudPrintJobResponse, mesh: bytes, on_finished: Callable[[], Any],
                        on_progress: Callable[[int], Any], on_error: Callable[[], Any]):
-        self._upload = ToolPathUploader(self._manager, print_job, mesh, on_finished, on_progress, on_error)
+        self._upload = ToolPathUploader(self._http, print_job, mesh, on_finished, on_progress, on_error)
         self._upload.start()
 
     # Requests a cluster to print the given print job.
@@ -93,8 +110,11 @@ class CloudApiClient:
     #  \param on_finished: The function to be called after the result is parsed.
     def requestPrint(self, cluster_id: str, job_id: str, on_finished: Callable[[CloudPrintResponse], Any]) -> None:
         url = "{}/clusters/{}/print/{}".format(self.CLUSTER_API_ROOT, cluster_id, job_id)
-        reply = self._manager.post(self._createEmptyRequest(url), b"")
+        self._http.post(url,
-        self._addCallback(reply, on_finished, CloudPrintResponse)
+                       scope = self._scope,
+                       data = b"",
+                       callback = self._parseCallback(on_finished, CloudPrintResponse),
+                       timeout = self.DEFAULT_REQUEST_TIMEOUT)
 
     ##  Send a print job action to the cluster for the given print job.
     #  \param cluster_id: The ID of the cluster.
@@ -104,7 +124,10 @@ class CloudApiClient:
                          data: Optional[Dict[str, Any]] = None) -> None:
         body = json.dumps({"data": data}).encode() if data else b""
         url = "{}/clusters/{}/print_jobs/{}/action/{}".format(self.CLUSTER_API_ROOT, cluster_id, cluster_job_id, action)
-        self._manager.post(self._createEmptyRequest(url), body)
+        self._http.post(url,
+                        scope = self._scope,
+                        data = body,
+                        timeout = self.DEFAULT_REQUEST_TIMEOUT)
 
     ##  We override _createEmptyRequest in order to add the user credentials.
     #   \param url: The URL to request
@@ -162,13 +185,12 @@ class CloudApiClient:
     #  \param on_finished: The callback in case the response is successful. Depending on the endpoint it will be either
     #       a list or a single item.
     #  \param model: The type of the model to convert the response to.
-    def _addCallback(self,
+    def _parseCallback(self,
-                     reply: QNetworkReply,
+                       on_finished: Union[Callable[[CloudApiClientModel], Any],
-                     on_finished: Union[Callable[[CloudApiClientModel], Any],
+                                          Callable[[List[CloudApiClientModel]], Any]],
-                                        Callable[[List[CloudApiClientModel]], Any]],
+                       model: Type[CloudApiClientModel],
-                     model: Type[CloudApiClientModel],
+                       on_error: Optional[Callable] = None) -> Callable[[QNetworkReply], None]:
-                     on_error: Optional[Callable] = None) -> None:
+        def parse(reply: QNetworkReply) -> None:
-        def parse() -> None:
             self._anti_gc_callbacks.remove(parse)
 
             # Don't try to parse the reply if we didn't get one
@@ -184,6 +206,4 @@ class CloudApiClient:
                 self._parseModels(response, on_finished, model)
 
         self._anti_gc_callbacks.append(parse)
-        reply.finished.connect(parse)
+        return parse
-        if on_error is not None:
-            reply.error.connect(on_error)

plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py

@@ -4,6 +4,7 @@ import os
 from typing import Dict, List, Optional
 
 from PyQt5.QtCore import QTimer
+from PyQt5.QtNetwork import QNetworkReply
 
 from UM import i18nCatalog
 from UM.Logger import Logger  # To log errors talking to the API.
@@ -43,7 +44,7 @@ class CloudOutputDeviceManager:
         # Persistent dict containing the remote clusters for the authenticated user.
         self._remote_clusters = {}  # type: Dict[str, CloudOutputDevice]
         self._account = CuraApplication.getInstance().getCuraAPI().account  # type: Account
-        self._api = CloudApiClient(self._account, on_error = lambda error: Logger.log("e", str(error)))
+        self._api = CloudApiClient(CuraApplication.getInstance(), on_error = lambda error: Logger.log("e", str(error)))
         self._account.loginStateChanged.connect(self._onLoginStateChanged)
 
         # Ensure we don't start twice.
@@ -127,7 +128,7 @@ class CloudOutputDeviceManager:
 
         self._onSyncFinished(True)
 
-    def _onGetRemoteClusterFailed(self):
+    def _onGetRemoteClusterFailed(self, reply: QNetworkReply, error: QNetworkReply.NetworkError):
         self._onSyncFinished(False)
 
     def _onDevicesDiscovered(self, clusters: List[CloudClusterResponse]) -> None:

plugins/UM3NetworkPrinting/src/Cloud/ToolPathUploader.py

@@ -1,11 +1,11 @@
 # Copyright (c) 2019 Ultimaker B.V.
 # !/usr/bin/env python
 # -*- coding: utf-8 -*-
-from PyQt5.QtCore import QUrl
+from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply
-from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply, QNetworkAccessManager
+from typing import Callable, Any, Tuple
-from typing import Optional, Callable, Any, Tuple, cast
 
 from UM.Logger import Logger
+from UM.TaskManagement.HttpRequestManager import HttpRequestManager
 
 from ..Models.Http.CloudPrintJobResponse import CloudPrintJobResponse
 
@@ -23,16 +23,16 @@ class ToolPathUploader:
     BYTES_PER_REQUEST = 256 * 1024
 
     ## Creates a mesh upload object.
-    #  \param manager: The network access manager that will handle the HTTP requests.
+    #  \param http: The HttpRequestManager that will handle the HTTP requests.
     #  \param print_job: The print job response that was returned by the cloud after registering the upload.
     #  \param data: The mesh bytes to be uploaded.
     #  \param on_finished: The method to be called when done.
     #  \param on_progress: The method to be called when the progress changes (receives a percentage 0-100).
     #  \param on_error: The method to be called when an error occurs.
-    def __init__(self, manager: QNetworkAccessManager, print_job: CloudPrintJobResponse, data: bytes,
+    def __init__(self, http: HttpRequestManager, print_job: CloudPrintJobResponse, data: bytes,
                  on_finished: Callable[[], Any], on_progress: Callable[[int], Any], on_error: Callable[[], Any]
                  ) -> None:
-        self._manager = manager
+        self._http = http
         self._print_job = print_job
         self._data = data
 
@@ -43,25 +43,12 @@ class ToolPathUploader:
         self._sent_bytes = 0
         self._retries = 0
         self._finished = False
-        self._reply = None  # type: Optional[QNetworkReply]
 
     ## Returns the print job for which this object was created.
     @property
     def printJob(self):
         return self._print_job
 
-    ##  Creates a network request to the print job upload URL, adding the needed content range header.
-    def _createRequest(self) -> QNetworkRequest:
-        request = QNetworkRequest(QUrl(self._print_job.upload_url))
-        request.setHeader(QNetworkRequest.ContentTypeHeader, self._print_job.content_type)
-
-        first_byte, last_byte = self._chunkRange()
-        content_range = "bytes {}-{}/{}".format(first_byte, last_byte - 1, len(self._data))
-        request.setRawHeader(b"Content-Range", content_range.encode())
-        Logger.log("i", "Uploading %s to %s", content_range, self._print_job.upload_url)
-
-        return request
-
     ## Determines the bytes that should be uploaded next.
     #  \return: A tuple with the first and the last byte to upload.
     def _chunkRange(self) -> Tuple[int, int]:
@@ -88,13 +75,23 @@ class ToolPathUploader:
             raise ValueError("The upload is already finished")
 
         first_byte, last_byte = self._chunkRange()
-        request = self._createRequest()
+        content_range = "bytes {}-{}/{}".format(first_byte, last_byte - 1, len(self._data))
 
-        # now send the reply and subscribe to the results
+        headers = {
-        self._reply = self._manager.put(request, self._data[first_byte:last_byte])
+            "Content-Type": self._print_job.content_type,
-        self._reply.finished.connect(self._finishedCallback)
+            "Content-Range": content_range
-        self._reply.uploadProgress.connect(self._progressCallback)
+        }
-        self._reply.error.connect(self._errorCallback)
+
+        Logger.log("i", "Uploading %s to %s", content_range, self._print_job.upload_url)
+
+        self._http.put(
+            url = self._print_job.upload_url,
+            headers_dict = headers,
+            data = self._data[first_byte:last_byte],
+            callback = self._finishedCallback,
+            error_callback = self._errorCallback,
+            upload_progress_callback = self._progressCallback
+        )
 
     ## Handles an update to the upload progress
     #  \param bytes_sent: The amount of bytes sent in the current request.
@@ -106,16 +103,14 @@ class ToolPathUploader:
             self._on_progress(int(total_sent / len(self._data) * 100))
 
     ## Handles an error uploading.
-    def _errorCallback(self) -> None:
+    def _errorCallback(self, reply: QNetworkReply, error: QNetworkReply.NetworkError) -> None:
-        reply = cast(QNetworkReply, self._reply)
         body = bytes(reply.readAll()).decode()
         Logger.log("e", "Received error while uploading: %s", body)
         self.stop()
         self._on_error()
 
     ## Checks whether a chunk of data was uploaded successfully, starting the next chunk if needed.
-    def _finishedCallback(self) -> None:
+    def _finishedCallback(self, reply: QNetworkReply) -> None:
-        reply = cast(QNetworkReply, self._reply)
         Logger.log("i", "Finished callback %s %s",
                    reply.attribute(QNetworkRequest.HttpStatusCodeAttribute), reply.url().toString())
 
@@ -133,7 +128,7 @@ class ToolPathUploader:
 
         # Http codes that are not to be retried are assumed to be errors.
         if status_code > 308:
-            self._errorCallback()
+            self._errorCallback(reply, None)
             return
 
         Logger.log("d", "status_code: %s, Headers: %s, body: %s", status_code,