众所周知，在PySide中，想要发送网络请求且不阻塞GUI线程，需要使用QNetworkAccessManager，但是这个东西用起来十分麻烦，需要写很多槽函数，而且不能必须要绑定在对象上，否则会报空指针。
这种写法非常不优雅，而且让代码变得十分复杂。因此在写项目的实践中，我写了这样一个库，可以简化网络请求，特此分享出来。
其使用方法可能有点像JS中的Promise，不过笔者精力和技术有限，没有深入了解过异步编程和JS Promise，所以可能不是最好的方法，但至少解决了笔者自己在项目中的问题。
废话少说，show you the code。
这是一个基本示例：
def test1():
app = QApplication()
print("不阻塞1")
(
QRequestReady(app)
.get("https://www.example.com")
.then(lambda content: print("do then 1"))
.catch(lambda error: print(f"error: {error.name}"))
.done()
)
print("不阻塞2")
(
QRequestReady(app)
.get("https://www.example.com")
.then(lambda content: print("do then 2"))
.catch(lambda error: print(f"error: {error.name}"))
.done()
)
app.exec()
输出：
不阻塞1
不阻塞2
do then 2
do then 1
可以看到这里的请求是完全不阻塞主线程的。
同时，你也可以在then中返回任何值，它会作为下一个then函数的参数传入，这让你可以很方便的把一些网络数据的处理逻辑给封装起来。
ps：QRequestReady.get方法会返回一个QPromise对象。
例如函数getResult中，你可能需要对请求api后返回的json进行一些过滤，你可以把这部分逻辑放到then中，然后返回这个QPromise对象，这样上级调用者就可以使用处理好的返回值进行其它操作，例如更新视图。
def test2():
"""测试能否正确传递返回值"""
app = QApplication()
def getResult():
return (
QRequestReady(app)
.get("https://www.example.com")
.then(lambda content: "some value")
)
getResult().then(lambda value: print(value)).catch(lambda error: print(error)).done()
app.exec()
输出：
some value
而在一些复杂的功能中，你可能需要针对请求的数据，再次发送请求。因此QPromise对于返回一个QPromise的then函数，会把之后的所有then函数都放到这个QPromise的then函数列表末尾，然后启动这个QPromise。
简单来说就是QPromise会自动处理串联请求。
阅读下面这个示例代码应该可以帮助你理解我的意思：
def test3():
# 测试在then中返回QPromise
app = QApplication()
request = QRequestReady(app)
# 创建一个请求，在then中返回访问baidu的promise，所以这里应该会打印出baidu网页
(
request.get("https://www.example.com")
.then(lambda _: request.get("https://www.baidu.com"))
.then(lambda content: print(content.decode()))
.done()
)
print("不阻塞")
app.exec()
输出：
不阻塞
<html>
<head>
<script>
location.replace(location.href.replace("https://","http://"));
</script>
</head>
<body>
<noscript><meta http-equiv="refresh" content="0;url=http://www.baidu.com/"></noscript>
</body>
</html>
由于时间和精力有限，这个库没有做到完善，很多地方可能会有bug，这里面也没有支持像js promise那样的raise和catch链。
这里将源码放出来分享给大家。
这些代码完全为笔者本人原创，采用BSD许可证开源。
如果你需要使用或者基于笔者的代码进行二次开发和修改，请务必留下这篇文章的链接与笔者的名字
from enum import Enum
from typing import Any, Callable, List, Self
from PySide6 import QtNetwork
from PySide6 import QtCore
from PySide6.QtCore import QObject
from PySide6.QtNetwork import QNetworkAccessManager
from PySide6.QtWidgets import QApplication
from features.common.tricks import interfaceMethod
class GlobalQNetworkAccessManager:
"""全局网络访问管理器
使用单例模式，每次new出来都是同一个实例。
"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = QNetworkAccessManager()
return cls._instance
class QRequestMode(Enum):
GET = 1
POST = 2
PUT = 3
class QPromise:
def __init__(self):
self.thenFuncList: List[Callable[..., Any]] = []
self.catchFunc: Callable[..., Any] | None = None
def then(self, func: Callable[..., Any]) -> Self:
self.thenFuncList.append(func)
return self
def catch(self, func: Callable[..., Any]) -> Self:
self.catchFunc = func
return self
@interfaceMethod
def done(self) -> Self: ...
class QRequestPromise(QObject, QPromise):
"""请求Promise
封装了Qt网络请求相关的调用，使用方法详见`QRequestReady`
"""
def __init__(self, parent: QObject, mode: QRequestMode, url: str):
super().__init__(parent)
self.url: str = url
self.mode = mode
self.lastFuncResult: Any = None
def then(self, func: Callable[..., Any]) -> Self:
self.thenFuncList.append(func)
return self
def catch(self, func: Callable[..., Any]) -> Self:
self.catchFunc = func
return self
def done(self) -> Self:
request = QtNetwork.QNetworkRequest(QtCore.QUrl(self.url))
match self.mode:
case QRequestMode.GET:
self.reply = GlobalQNetworkAccessManager().get(request)
case QRequestMode.POST:
raise NotImplementedError("QRequest POST not implemented")
case QRequestMode.PUT:
raise NotImplementedError("QRequest PUT not implemented")
self.reply.finished.connect(self.onFinished)
self.reply.errorOccurred.connect(self.onErrorOccurred)
return self
def onFinished(self) -> None:
if self.reply.error() != QtNetwork.QNetworkReply.NetworkError.NoError:
return
content = bytes(self.reply.readAll().data())
self.lastFuncResult: Any = content
for index, thenFunc in enumerate(self.thenFuncList):
self.lastFuncResult = thenFunc(self.lastFuncResult)
# 如果这个函数返回的也是QPromise，则不再执行后续的thenFunc
# 而是把后续的thenFunc全部放到这个QPromise中
# 等这个QPromise执行完毕，再由它去执行后续的thenFunc
if isinstance(self.lastFuncResult, QPromise):
self.lastFuncResult.thenFuncList.extend(self.thenFuncList[index + 1:])
self.lastFuncResult.done()
break
self.reply.deleteLater()
def one rrorOccurred(self, error: QtNetwork.QNetworkReply.NetworkError) -> None:
if self.catchFunc:
result = self.catchFunc(error)
self.reply.deleteLater()
class QDataPromise(QPromise):
def __init__(self, data: Any):
self.data = data
self.thenFuncList: List[Callable[..., Any]] = []
self.catchFunc: Callable[[QtNetwork.QNetworkReply.NetworkError], Any] | None = None
self.lastFuncResult = None
def then(self, func: Callable[..., Any]) -> Self:
self.thenFuncList.append(func)
return self
def catch(self, func: Callable[[QtNetwork.QNetworkReply.NetworkError], Any]) -> Self:
self.catchFunc = func
return self
def done(self) -> Self:
self.lastFuncResult: Any = self.data
for index, thenFunc in enumerate(self.thenFuncList):
self.lastFuncResult = thenFunc(self.lastFuncResult)
if isinstance(self.lastFuncResult, QPromise):
self.lastFuncResult.thenFuncList.extend(self.thenFuncList[index + 1:])
self.lastFuncResult.done()
break
return self
class QRequestReady(QObject):
"""Qt异步请求的封装
在QRequest之上封装了parent和mode参数，使得使用更加方便，且可以复用RequestReady对象，减少重复代码。
使用示例：
```
(
RequestReady(app)
.get("https://www.example.com")
.then(lambda content: print("do then"))
.catch(lambda error: print(f"error: {error.name}"))
.done()
)
```
其中then可以附加多个，前一个的返回值将作为后一个的参数。
由于网络请求需要时间，因此done函数调用后再附加then一般也可以正常运行。
但安全起见，建议封装功能函数中把自己的then附加上即可，catch和done都交给调用者来执行。
"""
def __init__(self, parent: QObject):
super().__init__(parent)
def get(self, url: str) -> QRequestPromise:
return QRequestPromise(self, QRequestMode.GET, url)
def post(self, url: str) -> QRequestPromise:
return QRequestPromise(self, QRequestMode.POST, url)
def put(self, url: str) -> QRequestPromise:
return QRequestPromise(self, QRequestMode.PUT, url)