如何在安静的调用中执行 PyQt5 应用程序

Question

上下文：

我有一个 Flask 应用程序为资源 POST /start 提供服务。要执行的逻辑涉及到 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 在后台线程上运行资源。 我考虑过的一个可能的解决方案是将 QApplication 作为 os 子进程运行，但并不理想。

问题：

是否可以将其保留在 Flask 应用程序中？

示例 PyQt 类：

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 示例

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)

Answer 1

Resource 在辅助线程中执行 post、get 等方法，以避免执行 flask 的线程不阻塞，因此 QApplication 运行在 Qt 禁止生成该错误的辅助线程中。

在这种情况下，解决方案是。

• 创建一个通过在主线程上运行的 QWebEnginePage 处理请求的类。

• 让烧瓶在辅助线程上运行，这样它就不会阻塞 Qt 事件循环。

• 通过 post 方法和处理请求的类之间的信号发送信息。

考虑到这一点，我已经实现了一个示例，您可以通过 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"}