如何在响应式调用上执行PyQt5应用程序 [英] How to execute PyQt5 application on a resful call
问题描述
上下文:
Context:
我有一个Flask应用程序，提供资源 POST/start .要执行的逻辑包括 PyQt5 QWebEnginePage ，该URL加载并返回有关该URL的某些数据.
I have a Flask application serving a resource POST /start. The logic to be executed involves a PyQt5 QWebEnginePage loading a URL and returning certain data about it.
问题:
Problem:
执行QApplication时(调用 app.exec _())，我得到警告:
When the QApplication is executed (calling app.exec_()) I get the warning:
WARNING: QApplication was not created in the main() thread.
，然后出现错误:
2019-07-17 13:06:19.461 Python[56513:5183122] *** Assertion failure in +[NSUndoManager _endTopLevelGroupings], /BuildRoot/Library/Caches/com.apple.xbs/Sources/Foundation/Foundation-1562/Foundation/Misc.subproj/NSUndoManager.m:361
2019-07-17 13:06:19.464 Python[56513:5183122] *** Terminating app due to uncaught exception 'NSInternalInconsistencyException', reason: '+[NSUndoManager(NSInternal) _endTopLevelGroupings] is only safe to invoke on the main thread.'
*** First throw call stack:
(
0 CoreFoundation 0x00007fff4e1abded __exceptionPreprocess + 256
1 libobjc.A.dylib 0x00007fff7a273720 objc_exception_throw + 48
...
...
122 libsystem_pthread.dylib 0x00007fff7b53826f _pthread_start + 70
123 libsystem_pthread.dylib 0x00007fff7b534415 thread_start + 13
)
libc++abi.dylib: terminating with uncaught exception of type NSException
Received signal 6
[0x00010a766de6]
[0x7fff7b52cb3d]
...
...
[0x000105a0de27]
[end of stack trace]
似乎QApplication总是需要在主线程上运行，但情况并非如此，因为flask在后台线程上运行资源.我考虑过的一种可能的解决方案是将QApplication作为os子进程运行，但并不理想.
It seems like the QApplication always needs to run on the main thread, which is not the case since flask runs resources on background threads. A possible solution i have considered is to run the QApplication as a os subprocess but is not ideal.
问题:
Question:
是否可以将其保留在Flask应用程序中?
Is it possible to keep it within the Flask app?
PyQt类示例:
Example PyQt class:
import sys
from PyQt5.QtWebEngineWidgets import QWebEnginePage
from PyQt5.QtWebEngineWidgets import QWebEngineProfile
from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QUrl
class PyQtWebClient(QWebEnginePage):
def __init__(self, url):
# Pointless variable for showcase purposes
self.total_runtime = None
self.app = QApplication(sys.argv)
self.profile = QWebEngineProfile()
# This is a sample to show the constructor I am actually using, my 'profile' is more complex than this
super().__init__(self.profile, None)
# Register callback to run when the page loads
self.loadFinished.connect(self._on_load_finished)
self.load(QUrl(url))
self.app.exec_()
def _on_load_finished(self):
self.total_runtime = 10
if __name__ == '__main__':
url = "https://www.example.com"
page = PyQtWebClient(url)
Flask示例app.py
Example Flask app.py
from flask import Flask
from flask_restful import Resource, Api
from lenomi import PyQtWebClient
app = Flask(__name__)
api = Api(app)
class TestPyqt5(Resource):
def post(self):
web = PyQtWebClient("http://www.example.com")
# At this point PyQtWebClient should have finished loading the url, and the process is done
print(web.total_runtime)
api.add_resource(TestPyqt5, "/pyqt")
if __name__ == '__main__':
app.run(debug=True)
推荐答案
资源在辅助线程中执行post，get等方法，以避免执行flask的线程不会阻塞，因此QApplication在该线程中运行Qt禁止生成该错误的辅助线程.
Resource executes the post, get, etc methods in secondary threads to avoid that the thread where flask is executed does not block, and therefore the QApplication is running in that secondary thread that Qt prohibits generating that error.
在这种情况下，解决方案是
In this case the solution is.
-
创建一个类，该类通过运行在主线程上的QWebEnginePage处理请求.
Create a class that handles requests through QWebEnginePage running on the main thread.
使烧瓶在辅助线程上运行，以免阻塞Qt事件循环.
Make the flask run on a secondary thread so that it does not block the Qt eventloop.
通过post方法和处理请求的类之间的信号发送信息.
Send the information through signals between the post method and the class that handles the requests.
考虑到这一点，我实现了一个示例，您可以在该示例中通过API来请求页面，获取该页面的HTML
Considering this I have implemented an example where you can make requests to pages via the API, obtaining the HTML of that page
lenomi.py
from functools import partial
from PyQt5 import QtCore, QtWebEngineWidgets
class Signaller(QtCore.QObject):
emitted = QtCore.pyqtSignal(object)
class PyQtWebClient(QtCore.QObject):
@QtCore.pyqtSlot(Signaller, str)
def get(self, signaller, url):
self.total_runtime = None
profile = QtWebEngineWidgets.QWebEngineProfile(self)
page = QtWebEngineWidgets.QWebEnginePage(profile, self)
wrapper = partial(self._on_load_finished, signaller)
page.loadFinished.connect(wrapper)
page.load(QtCore.QUrl(url))
@QtCore.pyqtSlot(Signaller, bool)
def _on_load_finished(self, signaller, ok):
page = self.sender()
if not isinstance(page, QtWebEngineWidgets.QWebEnginePage) or not ok:
signaller.emitted.emit(None)
return
self.total_runtime = 10
html = PyQtWebClient.download_html(page)
args = self.total_runtime, html
signaller.emitted.emit(args)
profile = page.profile()
page.deleteLater()
profile.deleteLater()
@staticmethod
def download_html(page):
html = ""
loop = QtCore.QEventLoop()
def callback(r):
nonlocal html
html = r
loop.quit()
page.toHtml(callback)
loop.exec_()
return html
app.py
import sys
import threading
from functools import partial
from flask import Flask
from flask_restful import Resource, Api, reqparse
from PyQt5 import QtCore, QtWidgets
from lenomi import PyQtWebClient, Signaller
app = Flask(__name__)
api = Api(app)
parser = reqparse.RequestParser()
class TestPyqt5(Resource):
def __init__(self, client):
self.m_client = client
def post(self):
parser.add_argument("url", type=str)
args = parser.parse_args()
url = args["url"]
if url:
total_runtime, html, error = 0, "", "not error"
def callback(loop, results=None):
if results is None:
nonlocal error
error = "Not load"
else:
nonlocal total_runtime, html
total_runtime, html = results
loop.quit()
signaller = Signaller()
loop = QtCore.QEventLoop()
signaller.emitted.connect(partial(callback, loop))
wrapper = partial(self.m_client.get, signaller, url)
QtCore.QTimer.singleShot(0, wrapper)
loop.exec_()
return {
"html": html,
"total_runtime": total_runtime,
"error": error,
}
qt_app = None
def main():
global qt_app
qt_app = QtWidgets.QApplication(sys.argv)
client = PyQtWebClient()
api.add_resource(
TestPyqt5, "/pyqt", resource_class_kwargs={"client": client}
)
threading.Thread(
target=app.run,
kwargs=dict(debug=False, use_reloader=False),
daemon=True,
).start()
return qt_app.exec_()
if __name__ == "__main__":
sys.exit(main())
curl http://localhost:5000/pyqt -d "url=https://www.example.com" -X POST
输出:
{"html": "<!DOCTYPE html><html><head>\n <title>Example Domain</title>\n\n <meta charset=\"utf-8\">\n <meta http-equiv=\"Content-type\" content=\"text/html; charset=utf-8\">\n <meta name=\"viewport\" content=\"width=device-width, initial-scale=1\">\n <style type=\"text/css\">\n body {\n background-color: #f0f0f2;\n margin: 0;\n padding: 0;\n font-family: \"Open Sans\", \"Helvetica Neue\", Helvetica, Arial, sans-serif;\n \n }\n div {\n width: 600px;\n margin: 5em auto;\n padding: 50px;\n background-color: #fff;\n border-radius: 1em;\n }\n a:link, a:visited {\n color: #38488f;\n text-decoration: none;\n }\n @media (max-width: 700px) {\n body {\n background-color: #fff;\n }\n div {\n width: auto;\n margin: 0 auto;\n border-radius: 0;\n padding: 1em;\n }\n }\n </style> \n</head>\n\n<body>\n<div>\n <h1>Example Domain</h1>\n <p>This domain is established to be used for illustrative examples in documents. You may use this\n domain in examples without prior coordination or asking for permission.</p>\n <p><a href=\"http://www.iana.org/domains/example\">More information...</a></p>\n</div>\n\n\n</body></html>", "total_runtime": 10, "error": "not error"}
这篇关于如何在响应式调用上执行PyQt5应用程序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
