如何在 restful 调用中执行 PyQt5 应用程序
How to execute PyQt5 application on a resful call
上下文：
我有一个提供资源 POST /start 的 Flask 应用程序。要执行的逻辑涉及 PyQt5 QWebEnginePage 加载 URL 并返回有关它的某些数据。
问题：
执行 QApplication 时（调用 app.exec_()）我收到警告：
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 运行s 资源在后台线程上。
我考虑过的 possible 解决方案是 运行 QApplication 作为 os 子进程，但并不理想。
问题：
是否可以将其保存在 Flask 应用程序中？os
示例 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)
示例烧瓶 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)
Resource 在副线程中执行post、get 等方法，以避免执行flask 的线程不阻塞，因此QApplication 运行ning 在那个副线程中Qt 禁止生成该错误。
在这种情况下，解决方案是。
创建一个class，在主线程上通过QWebEnginePage 运行ning处理请求。
将 flask 运行 放在辅助线程上，这样它就不会阻塞 Qt 事件循环。
通过信号在 post 方法和处理请求的 class 之间发送信息。
考虑到这一点，我实现了一个示例，您可以通过 API 向页面发出请求，获取该页面的 HTML
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"}
上下文：
我有一个提供资源 POST /start 的 Flask 应用程序。要执行的逻辑涉及 PyQt5 QWebEnginePage 加载 URL 并返回有关它的某些数据。
问题：
执行 QApplication 时（调用 app.exec_()）我收到警告：
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 运行s 资源在后台线程上。 我考虑过的 possible 解决方案是 运行 QApplication 作为 os 子进程，但并不理想。
问题：
是否可以将其保存在 Flask 应用程序中？os
示例 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)
示例烧瓶 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)
Resource 在副线程中执行post、get 等方法，以避免执行flask 的线程不阻塞，因此QApplication 运行ning 在那个副线程中Qt 禁止生成该错误。
在这种情况下，解决方案是。
创建一个class，在主线程上通过QWebEnginePage 运行ning处理请求。
将 flask 运行 放在辅助线程上，这样它就不会阻塞 Qt 事件循环。
通过信号在 post 方法和处理请求的 class 之间发送信息。
考虑到这一点，我实现了一个示例，您可以通过 API 向页面发出请求，获取该页面的 HTML
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"}