ZetCode

PyQt QNetworkAccessManager

最后修改于 2023 年 8 月 24 日

在本文中，我们将展示如何使用 QNetworkAccessManager 来发送请求和接收响应。

访问 Advanced PyQt5 电子书，阅读 PyQt6 教程，或列出所有 PyQt 教程。

QNetworkAccessManager

QNetworkAccessManager 允许应用程序发送网络请求并接收回复。QNetworkRequest 持有要通过网络管理器发送的请求，而 QNetworkReply 包含响应返回的数据和标头。

QNetworkAccessManager 有一个异步 API，这意味着它的方法总是立即返回，而不会等待它们完成。相反，当请求完成时，会发出一个信号。我们在附加到 finished 信号的方法中处理响应。

HTTP GET 请求

HTTP GET 方法请求指定资源的表示形式。

get_request.py 
#!/usr/bin/python

from PyQt6 import QtNetwork
from PyQt6.QtCore import QCoreApplication, QUrl
import sys


class Example:

    def __init__(self):

        self.doRequest()

    def doRequest(self):

        url = 'http://webcode.me'
        req = QtNetwork.QNetworkRequest(QUrl(url))

        self.nam = QtNetwork.QNetworkAccessManager()
        self.nam.finished.connect(self.handleResponse)
        self.nam.get(req)

    def handleResponse(self, reply):

        er = reply.error()

        if er == QtNetwork.QNetworkReply.NetworkError.NoError:

            bytes_string = reply.readAll()
            print(str(bytes_string, 'utf-8'))

        else:
            print("Error occured: ", er)
            print(reply.errorString())

        QCoreApplication.quit()


def main():

    app = QCoreApplication([])
    ex = Example()
    sys.exit(app.exec())


if __name__ == '__main__':
    main()

该示例检索指定网页的 HTML 代码。

url = 'http://webcode.me'
req = QtNetwork.QNetworkRequest(QUrl(url))

我们使用 QNetworkRequest 向指定的 URL 发送请求。

self.nam = QtNetwork.QNetworkAccessManager()
self.nam.finished.connect(self.handleResponse)
self.nam.get(req)

创建一个 QNetworkAccessManager 对象。当请求完成时，调用 handleResponse 方法。请求通过 get 方法发出。

def handleResponse(self, reply):

    er = reply.error()

    if er == QtNetwork.QNetworkReply.NetworkError.NoError:

        bytes_string = reply.readAll()
        print(str(bytes_string, 'utf-8'))

    else:
        print("Error occured: ", er)
        print(reply.errorString())

    QCoreApplication.quit()

handleResponse 接收一个 QNetworkReply 对象。它包含已发送请求的数据和标头。如果网络回复中没有错误，我们使用 readAll 方法读取所有数据；否则我们打印一条错误消息。errorString 返回对最后发生的错误的人类可读描述。readAll 返回 QByteArray 格式的数据，需要进行解码。

$ ./get_request.py 
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>My html page</title>
</head>
<body>

    <p>
        Today is a beautiful day. We go swimming and fishing.
    </p>
    
    <p>
         Hello there. How are you?
    </p>
    
</body>
</html>

HTTP POST 请求

HTTP POST 方法向服务器发送数据。请求正文的类型由 Content-Type 标头指示。POST 请求通常通过 HTML 表单发送。请求中发送的数据可以用不同的方式编码；在 application/x-www-form-urlencoded 中，值被编码为由 '&' 分隔的键值元组，键和值之间用 '=' 连接。非字母数字字符会进行百分比编码。multipart/form-data 用于二进制数据和文件上传。

post_request.py 
#!/usr/bin/python


from PyQt6 import QtNetwork
from PyQt6 import QtCore
import sys, json

class Example:

    def __init__(self):

        self.doRequest()


    def doRequest(self):

        data = QtCore.QByteArray()
        data.append(b'name=Peter&amp;')
        data.append(b'age=34')

        url = 'https://httpbin.org/post'
        req = QtNetwork.QNetworkRequest(QtCore.QUrl(url))
        req.setHeader(QtNetwork.QNetworkRequest.KnownHeaders.ContentTypeHeader,
            'application/x-www-form-urlencoded')

        self.nam = QtNetwork.QNetworkAccessManager()
        self.nam.finished.connect(self.handleResponse)
        self.nam.post(req, data)


    def handleResponse(self, reply):

        er = reply.error()

        if er == QtNetwork.QNetworkReply.NetworkError.NoError:

            bytes_string = reply.readAll()

            json_ar = json.loads(str(bytes_string, 'utf-8'))
            data = json_ar['form']

            print(f"Name: {data['name']}")
            print(f"Age: {data['age']}")
            print()

        else:
            print('Error occurred: ', er)
            print(reply.errorString())

        QtCore.QCoreApplication.quit()


def main():

    app = QtCore.QCoreApplication([])
    ex = Example()
    sys.exit(app.exec())


if __name__ == '__main__':
    main()

该示例向测试网站 https://httpbin.org/post 发送一个 post 请求，该网站会以 JSON 格式将数据返回。

data = QtCore.QByteArray()
data.append(b'name=Peter&amp;')
data.append(b'age=34')

根据规范，我们对发送的数据在 QByteArray 中进行编码。

url = 'https://httpbin.org/post'
req = QtNetwork.QNetworkRequest(QtCore.QUrl(url))
req.setHeader(QtNetwork.QNetworkRequest.KnownHeaders.ContentTypeHeader,
    'application/x-www-form-urlencoded')

我们指定 application/x-www-form-urlencoded 编码类型。

bytes_string = reply.readAll()

json_ar = json.loads(str(bytes_string, 'utf-8'))
data = json_ar['form']

print(f"Name: {data['name']}")
print(f"Age: {data['age']}")
print()

在处理方法中，我们读取响应数据并对其进行解码。我们使用内置的 json 模块来提取已提交的数据。

$ ./post_request.py
Name: Peter
Age: 34

QNetworkAccessManager 身份验证

每当最终服务器在传递所请求的内容之前请求身份验证时，就会发出 authenticationRequired 信号。

authentication.py 
#!/usr/bin/python

from PyQt6 import QtCore, QtNetwork
import sys, json


class Example:

    def __init__(self):

        self.doRequest()


    def doRequest(self):

        self.auth = 0

        url = 'https://httpbin.org/basic-auth/user7/passwd7'
        req = QtNetwork.QNetworkRequest(QtCore.QUrl(url))

        self.nam = QtNetwork.QNetworkAccessManager()
        self.nam.authenticationRequired.connect(self.authenticate)
        self.nam.finished.connect(self.handleResponse)
        self.nam.get(req)


    def authenticate(self, reply, auth):

        print('Authenticating')

        self.auth += 1

        if self.auth >= 3:
            reply.abort()

        auth.setUser('user7')
        auth.setPassword('passwd7')


    def handleResponse(self, reply):

        er = reply.error()

        if er == QtNetwork.QNetworkReply.NetworkError.NoError:

            bytes_string = reply.readAll()

            data = json.loads(str(bytes_string, 'utf-8'))

            print(f"Authenticated: {data['authenticated']}")
            print(f"User: {data['user']}")

            print()

        else:
            print('Error occurred: ', er)
            print(reply.errorString())

        QtCore.QCoreApplication.quit()


def main():

    app = QtCore.QCoreApplication([])
    ex = Example()
    sys.exit(app.exec())


if __name__ == '__main__':
    main()

在该示例中，我们使用 https://httpbin.org 网站来展示如何使用 QNetworkAccessManager 进行身份验证。

self.nam.authenticationRequired.connect(self.authenticate)

我们将 authenticationRequired 信号连接到 authenticate 方法。

def authenticate(self, reply, auth):

    print('Authenticating')
    ...

authenticate 方法的第三个参数是 QAuthenticator，它用于传递所需的身份验证信息。

self.auth += 1

if self.auth >= 3:
    reply.abort()

如果身份验证失败，QNetworkAccessManager 会持续发出 authenticationRequired 信号。我们在三次失败尝试后中止该过程。

auth.setUser('user7')
auth.setPassword('passwd7')

我们将用户和密码设置到 QAuthenticator 中。

bytes_string = reply.readAll()

data = json.loads(str(bytes_string, 'utf-8'))

print(f"Authenticated: {data['authenticated']}")
print(f"User: {data['user']}")

print()

https://httpbin.org 以 JSON 数据响应，其中包含用户名和一个指示身份验证是否成功的布尔值。

$ ./authentication.py
Authenticating
Authenticated: True
User: user7

使用 QNetworkAccessManager 获取网站图标 (favicon)

网站图标 (favicon) 是与特定网站相关联的小图标。在下面的示例中，我们将从一个网站下载一个网站图标。

fetch_icon.py 
#!/usr/bin/python

from PyQt6 import QtCore, QtNetwork
import sys


class Example:

    def __init__(self):

        self.doRequest()

    def doRequest(self):

        url = 'http://webcode.me/favicon.ico'
        req = QtNetwork.QNetworkRequest(QtCore.QUrl(url))

        self.nam = QtNetwork.QNetworkAccessManager()
        self.nam.finished.connect(self.handleResponse)
        self.nam.get(req)

    def handleResponse(self, reply):

        er = reply.error()

        if er == QtNetwork.QNetworkReply.NetworkError.NoError:

            data = reply.readAll()
            self.saveFile(data)

        else:
            print('Error occured: ', er)
            print(reply.errorString())

        QtCore.QCoreApplication.quit()

    def saveFile(self, data):

        f = open('favicon.ico', 'wb')

        with f:

            f.write(data)


def main():

    app = QtCore.QCoreApplication([])
    ex = Example()
    sys.exit(app.exec())


if __name__ == '__main__':
    main()

该代码示例下载了谷歌的网站图标。

self.nam.get(req)

我们使用 get 方法下载图标。

data = reply.readAll()
self.saveFile(data)

在 handleResponse 方法中，我们读取数据并将其保存到文件中。

def saveFile(self, data):

    f = open('favicon.ico', 'wb')

    with f:

        f.write(data)

图像数据在 saveFile 方法中被保存在磁盘上。

在本文中，我们学习了如何使用 QNetworkAccessManager。

作者

我的名字是 Jan Bodnar，我是一名充满热情的程序员，拥有丰富的编程经验。我从 2007 年开始撰写编程文章。迄今为止，我已经撰写了超过 1400 篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 PyQt 教程。