From f3c66c31898a4167def7b2b89860e5495621b84c Mon Sep 17 00:00:00 2001 From: Nino van Hooff Date: Fri, 15 May 2020 11:28:17 +0200 Subject: [PATCH] Refactor CloudApiClient (and ToolpathUploader) to use HttpRequestManager Has the benefit of a more unified Http request management + timeouts CURA-7290 --- .../src/Cloud/CloudApiClient.py | 72 ++++++++++++------- .../src/Cloud/CloudOutputDeviceManager.py | 5 +- .../src/Cloud/ToolPathUploader.py | 55 +++++++------- 3 files changed, 74 insertions(+), 58 deletions(-) diff --git a/plugins/UM3NetworkPrinting/src/Cloud/CloudApiClient.py b/plugins/UM3NetworkPrinting/src/Cloud/CloudApiClient.py index 1c9670d87f..b45f549b3c 100644 --- a/plugins/UM3NetworkPrinting/src/Cloud/CloudApiClient.py +++ b/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._account = account + self._app = app + 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._addCallback(reply, on_finished, CloudClusterResponse, failed) + self._http.get(url, + 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._addCallback(reply, on_finished, CloudClusterStatus) + self._http.get(url, + 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()}) - reply = self._manager.put(self._createEmptyRequest(url), body.encode()) - self._addCallback(reply, on_finished, CloudPrintJobResponse) + data = json.dumps({"data": request.toDict()}).encode() + + 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._addCallback(reply, on_finished, CloudPrintResponse) + self._http.post(url, + 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, - reply: QNetworkReply, - on_finished: Union[Callable[[CloudApiClientModel], Any], - Callable[[List[CloudApiClientModel]], Any]], - model: Type[CloudApiClientModel], - on_error: Optional[Callable] = None) -> None: - def parse() -> None: + def _parseCallback(self, + on_finished: Union[Callable[[CloudApiClientModel], Any], + Callable[[List[CloudApiClientModel]], Any]], + model: Type[CloudApiClientModel], + on_error: Optional[Callable] = None) -> Callable[[QNetworkReply], None]: + def parse(reply: QNetworkReply) -> 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) - if on_error is not None: - reply.error.connect(on_error) + return parse diff --git a/plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py b/plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py index 8a87438b0e..2d941281c7 100644 --- a/plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py +++ b/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: diff --git a/plugins/UM3NetworkPrinting/src/Cloud/ToolPathUploader.py b/plugins/UM3NetworkPrinting/src/Cloud/ToolPathUploader.py index 6aa341c0e5..6317dff347 100644 --- a/plugins/UM3NetworkPrinting/src/Cloud/ToolPathUploader.py +++ b/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, QNetworkAccessManager -from typing import Optional, Callable, Any, Tuple, cast +from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply +from typing import Callable, Any, Tuple 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 - self._reply = self._manager.put(request, self._data[first_byte:last_byte]) - self._reply.finished.connect(self._finishedCallback) - self._reply.uploadProgress.connect(self._progressCallback) - self._reply.error.connect(self._errorCallback) + headers = { + "Content-Type": self._print_job.content_type, + "Content-Range": content_range + } + + 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: - reply = cast(QNetworkReply, self._reply) + def _errorCallback(self, reply: QNetworkReply, error: QNetworkReply.NetworkError) -> None: 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: - reply = cast(QNetworkReply, self._reply) + def _finishedCallback(self, reply: QNetworkReply) -> None: 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,