|
1 # -*- coding: utf-8 -*- |
|
2 |
|
3 # Copyright (c) 2014 - 2022 Detlev Offenbach <detlev@die-offenbachs.de> |
|
4 # |
|
5 |
|
6 """ |
|
7 Module implementing a synchronous network request handler for translation |
|
8 requests. |
|
9 """ |
|
10 |
|
11 from PyQt6.QtCore import QObject, QEventLoop, QByteArray |
|
12 from PyQt6.QtNetwork import ( |
|
13 QNetworkAccessManager, QNetworkRequest, QNetworkReply |
|
14 ) |
|
15 |
|
16 from EricNetwork.EricNetworkProxyFactory import proxyAuthenticationRequired |
|
17 |
|
18 |
|
19 class TranslatorRequest(QObject): |
|
20 """ |
|
21 Class implementing a synchronous network request handler for translation |
|
22 requests. |
|
23 """ |
|
24 def __init__(self, parent=None): |
|
25 """ |
|
26 Constructor |
|
27 |
|
28 @param parent reference to the parent object |
|
29 @type QObject |
|
30 """ |
|
31 super().__init__(parent) |
|
32 |
|
33 self.__contentTypes = { |
|
34 "form": b"application/x-www-form-urlencoded", |
|
35 "json": b"application/json", |
|
36 } |
|
37 |
|
38 self.__networkManager = QNetworkAccessManager(self) |
|
39 self.__networkManager.proxyAuthenticationRequired.connect( |
|
40 proxyAuthenticationRequired) |
|
41 |
|
42 self.__loop = QEventLoop() |
|
43 self.__networkManager.finished.connect(self.__loop.quit) |
|
44 |
|
45 def get(self, requestUrl, extraHeaders=None): |
|
46 """ |
|
47 Public method to issue a GET request. |
|
48 |
|
49 @param requestUrl URL of the request |
|
50 @type QUrl |
|
51 @param extraHeaders list of tuples of additional headers giving |
|
52 header name and header value |
|
53 @type tuple of (bytes, bytes) |
|
54 @return server response or an error message and a success flag |
|
55 @rtype tuple of (QByteArray or str, bool) |
|
56 """ |
|
57 request = QNetworkRequest(requestUrl) |
|
58 if extraHeaders: |
|
59 for name, value in extraHeaders: |
|
60 request.setRawHeader(name, value) |
|
61 reply = self.__networkManager.get(request) |
|
62 if not self.__loop.isRunning(): |
|
63 self.__loop.exec() |
|
64 if reply.error() != QNetworkReply.NetworkError.NoError: |
|
65 return reply.errorString(), False |
|
66 else: |
|
67 return reply.readAll(), True |
|
68 |
|
69 def post(self, requestUrl, requestData, dataType="form", |
|
70 extraHeaders=None): |
|
71 """ |
|
72 Public method to issue a POST request. |
|
73 |
|
74 @param requestUrl URL of the request |
|
75 @type QUrl |
|
76 @param requestData data of the request |
|
77 @type QByteArray |
|
78 @param dataType type of the request data |
|
79 @type str |
|
80 @param extraHeaders list of tuples of additional headers giving |
|
81 header name and header value |
|
82 @type list of tuple of (bytes, bytes) |
|
83 @return tuple of server response (string) and flag indicating |
|
84 success (boolean) |
|
85 @rtype tuple of (str, bool) |
|
86 """ |
|
87 request = QNetworkRequest(requestUrl) |
|
88 request.setRawHeader(b"User-Agent", |
|
89 b"Mozilla/5.0") |
|
90 request.setRawHeader(b"Content-Type", |
|
91 self.__contentTypes[dataType]) |
|
92 request.setRawHeader(b"Content-Length", |
|
93 QByteArray.number(requestData.size())) |
|
94 if extraHeaders: |
|
95 for name, value in extraHeaders: |
|
96 request.setRawHeader(name, value) |
|
97 request.setUrl(requestUrl) |
|
98 reply = self.__networkManager.post(request, requestData) |
|
99 if not self.__loop.isRunning(): |
|
100 self.__loop.exec() |
|
101 if reply.error() != QNetworkReply.NetworkError.NoError: |
|
102 return reply.errorString(), False |
|
103 else: |
|
104 return str(reply.readAll(), "utf-8", "replace"), True |