diff options
Diffstat (limited to 'plugins/DigitalLibrary/src/DigitalFactoryApiClient.py')
| -rw-r--r-- | plugins/DigitalLibrary/src/DigitalFactoryApiClient.py | 317 |
1 files changed, 317 insertions, 0 deletions
diff --git a/plugins/DigitalLibrary/src/DigitalFactoryApiClient.py b/plugins/DigitalLibrary/src/DigitalFactoryApiClient.py new file mode 100644 index 0000000000..4342d2623e --- /dev/null +++ b/plugins/DigitalLibrary/src/DigitalFactoryApiClient.py @@ -0,0 +1,317 @@ +# Copyright (c) 2021 Ultimaker B.V. +# Cura is released under the terms of the LGPLv3 or higher. + +import json +from json import JSONDecodeError +import re +from time import time +from typing import List, Any, Optional, Union, Type, Tuple, Dict, cast, TypeVar, Callable + +from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest + +from UM.Logger import Logger +from UM.TaskManagement.HttpRequestManager import HttpRequestManager +from UM.TaskManagement.HttpRequestScope import JsonDecoratorScope +from cura.CuraApplication import CuraApplication +from cura.UltimakerCloud import UltimakerCloudConstants +from cura.UltimakerCloud.UltimakerCloudScope import UltimakerCloudScope +from .DFPrintJobUploadResponse import DFPrintJobUploadResponse +from .BaseModel import BaseModel +from .CloudError import CloudError +from .DFFileUploader import DFFileUploader +from .DFLibraryFileUploadRequest import DFLibraryFileUploadRequest +from .DFLibraryFileUploadResponse import DFLibraryFileUploadResponse +from .DFPrintJobUploadRequest import DFPrintJobUploadRequest +from .DigitalFactoryFileResponse import DigitalFactoryFileResponse +from .DigitalFactoryProjectResponse import DigitalFactoryProjectResponse +from .PaginationManager import PaginationManager + +CloudApiClientModel = TypeVar("CloudApiClientModel", bound=BaseModel) +"""The generic type variable used to document the methods below.""" + + +class DigitalFactoryApiClient: + # The URL to access the digital factory. + ROOT_PATH = UltimakerCloudConstants.CuraCloudAPIRoot + 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[[Any], None]] + + def __init__(self, application: CuraApplication, on_error: Callable[[List[CloudError]], None], projects_limit_per_page: Optional[int] = None) -> None: + """Initializes a new digital factory API client. + + :param application: + :param on_error: The callback to be called whenever we receive errors from the server. + """ + super().__init__() + self._application = application + self._account = application.getCuraAPI().account + self._scope = JsonDecoratorScope(UltimakerCloudScope(application)) + self._http = HttpRequestManager.getInstance() + self._on_error = on_error + self._file_uploader = None # type: Optional[DFFileUploader] + + self._projects_pagination_mgr = PaginationManager(limit = projects_limit_per_page) if projects_limit_per_page else None # type: Optional[PaginationManager] + + def getProject(self, library_project_id: str, on_finished: Callable[[DigitalFactoryProjectResponse], Any], failed: Callable) -> None: + """ + Retrieves a digital factory project by its library project id. + + :param library_project_id: The id of the library project + :param on_finished: The function to be called after the result is parsed. + :param failed: The function to be called if the request fails. + """ + url = "{}/projects/{}".format(self.CURA_API_ROOT, library_project_id) + + self._http.get(url, + scope = self._scope, + callback = self._parseCallback(on_finished, DigitalFactoryProjectResponse, failed), + error_callback = failed, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def getProjectsFirstPage(self, on_finished: Callable[[List[DigitalFactoryProjectResponse]], Any], failed: Callable) -> None: + """ + Retrieves digital factory projects for the user that is currently logged in. + + If a projects pagination manager exists, then it attempts to get the first page of the paginated projects list, + according to the limit set in the pagination manager. If there is no projects pagination manager, this function + leaves the project limit to the default set on the server side (999999). + + :param on_finished: The function to be called after the result is parsed. + :param failed: The function to be called if the request fails. + """ + url = "{}/projects".format(self.CURA_API_ROOT) + if self._projects_pagination_mgr: + self._projects_pagination_mgr.reset() # reset to clear all the links and response metadata + url += "?limit={}".format(self._projects_pagination_mgr.limit) + + self._http.get(url, + scope = self._scope, + callback = self._parseCallback(on_finished, DigitalFactoryProjectResponse, failed, pagination_manager = self._projects_pagination_mgr), + error_callback = failed, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def getMoreProjects(self, + on_finished: Callable[[List[DigitalFactoryProjectResponse]], Any], + failed: Callable) -> None: + """Retrieves the next page of the paginated projects list from the API, provided that there is any. + + :param on_finished: The function to be called after the result is parsed. + :param failed: The function to be called if the request fails. + """ + + if self.hasMoreProjectsToLoad(): + url = self._projects_pagination_mgr.links.next_page + self._http.get(url, + scope = self._scope, + callback = self._parseCallback(on_finished, DigitalFactoryProjectResponse, failed, pagination_manager = self._projects_pagination_mgr), + error_callback = failed, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + else: + Logger.log("d", "There are no more projects to load.") + + def hasMoreProjectsToLoad(self) -> bool: + """ + Determines whether the client can get more pages of projects list from the API. + + :return: Whether there are more pages in the projects list available to be retrieved from the API. + """ + return self._projects_pagination_mgr and self._projects_pagination_mgr.links and self._projects_pagination_mgr.links.next_page is not None + + def getListOfFilesInProject(self, library_project_id: str, on_finished: Callable[[List[DigitalFactoryFileResponse]], Any], failed: Callable) -> None: + """Retrieves the list of files contained in the project with library_project_id from the Digital Factory Library. + + :param library_project_id: The id of the digital factory library project in which the files are included + :param on_finished: The function to be called after the result is parsed. + :param failed: The function to be called if the request fails. + """ + + url = "{}/projects/{}/files".format(self.CURA_API_ROOT, library_project_id) + self._http.get(url, + scope = self._scope, + callback = self._parseCallback(on_finished, DigitalFactoryFileResponse, failed), + error_callback = failed, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def _parseCallback(self, + on_finished: Union[Callable[[CloudApiClientModel], Any], + Callable[[List[CloudApiClientModel]], Any]], + model: Type[CloudApiClientModel], + on_error: Optional[Callable] = None, + pagination_manager: Optional[PaginationManager] = None) -> Callable[[QNetworkReply], None]: + + """ + Creates a callback function so that it includes the parsing of the response into the correct model. + The callback is added to the 'finished' signal of the reply. If a paginated request was made and a pagination + manager is given, the pagination metadata will be held there. + + :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. + :param on_error: The callback in case the response is ... less successful. + :param pagination_manager: Holds the pagination links and metadata contained in paginated responses. + If no pagination manager is provided, the pagination metadata is ignored. + """ + + def parse(reply: QNetworkReply) -> None: + + self._anti_gc_callbacks.remove(parse) + + # Don't try to parse the reply if we didn't get one + if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) is None: + if on_error is not None: + on_error() + return + + status_code, response = self._parseReply(reply) + if status_code >= 300 and on_error is not None: + on_error() + else: + self._parseModels(response, on_finished, model, pagination_manager = pagination_manager) + + self._anti_gc_callbacks.append(parse) + return parse + + @staticmethod + def _parseReply(reply: QNetworkReply) -> Tuple[int, Dict[str, Any]]: + """Parses the given JSON network reply into a status code and a dictionary, handling unexpected errors as well. + + :param reply: The reply from the server. + :return: A tuple with a status code and a dictionary. + """ + + status_code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) + try: + response = bytes(reply.readAll()).decode() + return status_code, json.loads(response) + except (UnicodeDecodeError, JSONDecodeError, ValueError) as err: + error = CloudError(code = type(err).__name__, title = str(err), http_code = str(status_code), + id = str(time()), http_status = "500") + Logger.logException("e", "Could not parse the stardust response: %s", error.toDict()) + return status_code, {"errors": [error.toDict()]} + + def _parseModels(self, + response: Dict[str, Any], + on_finished: Union[Callable[[CloudApiClientModel], Any], + Callable[[List[CloudApiClientModel]], Any]], + model_class: Type[CloudApiClientModel], + pagination_manager: Optional[PaginationManager] = None) -> None: + """Parses the given models and calls the correct callback depending on the result. + + :param response: The response from the server, after being converted to a dict. + :param on_finished: The callback in case the response is successful. + :param model_class: The type of the model to convert the response to. It may either be a single record or a list. + :param pagination_manager: Holds the pagination links and metadata contained in paginated responses. + If no pagination manager is provided, the pagination metadata is ignored. + """ + + if "data" in response: + data = response["data"] + if "meta" in response and pagination_manager: + pagination_manager.setResponseMeta(response["meta"]) + if "links" in response and pagination_manager: + pagination_manager.setLinks(response["links"]) + if isinstance(data, list): + results = [model_class(**c) for c in data] # type: List[CloudApiClientModel] + on_finished_list = cast(Callable[[List[CloudApiClientModel]], Any], on_finished) + on_finished_list(results) + else: + result = model_class(**data) # type: CloudApiClientModel + on_finished_item = cast(Callable[[CloudApiClientModel], Any], on_finished) + on_finished_item(result) + elif "errors" in response: + self._on_error([CloudError(**error) for error in response["errors"]]) + else: + Logger.log("e", "Cannot find data or errors in the cloud response: %s", response) + + def requestUpload3MF(self, request: DFLibraryFileUploadRequest, + on_finished: Callable[[DFLibraryFileUploadResponse], Any], + on_error: Optional[Callable[["QNetworkReply", "QNetworkReply.NetworkError"], None]] = None) -> None: + + """Requests the Digital Factory to register the upload of a file in a library project. + + :param request: The request object. + :param on_finished: The function to be called after the result is parsed. + :param on_error: The callback in case the request fails. + """ + + url = "{}/files/upload".format(self.CURA_API_ROOT) + data = json.dumps({"data": request.toDict()}).encode() + + self._http.put(url, + scope = self._scope, + data = data, + callback = self._parseCallback(on_finished, DFLibraryFileUploadResponse), + error_callback = on_error, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def requestUploadUFP(self, request: DFPrintJobUploadRequest, + on_finished: Callable[[DFPrintJobUploadResponse], Any], + on_error: Optional[Callable[["QNetworkReply", "QNetworkReply.NetworkError"], None]] = None) -> None: + """Requests the Digital Factory to register the upload of a file in a library project. + + :param request: The request object. + :param on_finished: The function to be called after the result is parsed. + :param on_error: The callback in case the request fails. + """ + + url = "{}/jobs/upload".format(self.CURA_API_ROOT) + data = json.dumps({"data": request.toDict()}).encode() + + self._http.put(url, + scope = self._scope, + data = data, + callback = self._parseCallback(on_finished, DFPrintJobUploadResponse), + error_callback = on_error, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def uploadExportedFileData(self, + df_file_upload_response: Union[DFLibraryFileUploadResponse, DFPrintJobUploadResponse], + mesh: bytes, + on_finished: Callable[[str], Any], + on_success: Callable[[str], Any], + on_progress: Callable[[str, int], Any], + on_error: Callable[[str, "QNetworkReply", "QNetworkReply.NetworkError"], Any]) -> None: + + """Uploads an exported file (in bytes) to the Digital Factory Library. + + :param df_file_upload_response: The response received after requesting an upload with `self.requestUpload`. + :param mesh: The mesh data (in bytes) to be uploaded. + :param on_finished: The function to be called after the upload has finished. Called both after on_success and on_error. + It receives the name of the file that has finished uploading. + :param on_success: The function to be called if the upload was successful. + It receives the name of the file that was uploaded successfully. + :param on_progress: A function to be called during upload progress. It receives a percentage (0-100). + It receives the name of the file for which the upload progress should be updated. + :param on_error: A function to be called if the upload fails. + It receives the name of the file that produced errors during the upload process. + """ + + self._file_uploader = DFFileUploader(self._http, df_file_upload_response, mesh, on_finished, on_success, on_progress, on_error) + self._file_uploader.start() + + def createNewProject(self, project_name: str, on_finished: Callable[[CloudApiClientModel], Any], on_error: Callable) -> None: + """ Create a new project in the Digital Factory. + + :param project_name: Name of the new to be created project. + :param on_finished: The function to be called after the result is parsed. + :param on_error: The function to be called if anything goes wrong. + """ + + display_name = re.sub(r"[^a-zA-Z0-9- ./™®ö+']", " ", project_name) + Logger.log("i", "Attempt to create new DF project '{}'.".format(display_name)) + + url = "{}/projects".format(self.CURA_API_ROOT) + data = json.dumps({"data": {"display_name": display_name}}).encode() + self._http.put(url, + scope = self._scope, + data = data, + callback = self._parseCallback(on_finished, DigitalFactoryProjectResponse), + error_callback = on_error, + timeout = self.DEFAULT_REQUEST_TIMEOUT) + + def clear(self) -> None: + self._projects_pagination_mgr.reset() |