详解如何优雅的用PyQt访问http

# coding: utf-8
from typing import any

import requests
from pydantic import field, basemodel


class responsemodel(basemodel):
    status: bool = field(true, description="响应状态")
    message: str = field('请求成功', description="响应信息")
    error: str = field('', description="错误信息")
    response: requests.response = field(none, description="响应对象")
    result: any = field({}, description="响应数据")

    class config:
        arbitrary_types_allowed = true

# coding: utf-8
from pyside6.qtcore import qthread, signal, slot

from common.models import responsemodel
from loguru import logger


class requestthread(qthread):
    modelchanged = signal(responsemodel)

    def __init__(self, parent=none):
        super().__init__(parent)
        self.request = none

    def run(self):
        if not self.request:
            return
        try:
            result = self.request()
            self.modelchanged.emit(
                responsemodel(
                    result=result,
                )
            )
        except exception as e:
            logger.error(e)
            self.modelchanged.emit(
                responsemodel(
                    status=false,
                    message='请求失败',
                    error=e,
                )
            )

    def setrequest(self, request):
        self.request = request
        if not self.isrunning():
            self.start()

import requests
from pyside6.qtcore import qcoreapplication

from common.models import responsemodel
from request_thread import requestthread


def baidu_request():
    response = requests.get('http://www.baidu.com')
    response.encoding = 'utf-8'
    return response.text


def response_handler(response_model: responsemodel):
    if response_model.status:
        # todo: 当请求正确时处理逻辑
        print(response_model.result)
        pass
    else:
        # todo: 当请求错误时处理逻辑
        print(response_model.message, response_model.error)


if __name__ == '__main__':
    app = qcoreapplication([])
    thread = requestthread()
    thread.modelchanged.connect(response_handler)
    thread.finished.connect(app.quit)
    thread.setrequest(baidu_request)
    app.exec()

# coding: utf-8
from typing import union, any

from pydantic import basemodel, field


class requestmodel(basemodel):
    method: str = field('get', description='请求方法，如 get、post、put、delete')
    url: str = field(..., description='请求的 url 地址')
    params: dict = field(none, description='请求参数，如 get 请求时附带的参数')
    data: dict = field(none, description='请求数据，如 post 请求时提交的数据')
    json_: dict = field(none, description='请求数据，如 post 请求时提交的 json 数据', alias='json')
    headers: dict = field(none, description='请求头，如 content-type、user-agent 等')
    cookies: dict = field(none, description='请求 cookies，如登录后获取的 cookie')
    files: dict = field(none, description='上传的文件，如 post 请求时上传的文件')
    auth: union[tuple, list] = field(none, description='http 认证，如 basic 认证')
    timeout: int = field(none, description='请求超时时间，单位为秒')
    allow_redirects: bool = field(true, description='是否允许重定向')
    proxies: dict = field(none, description='代理设置')
    hooks: any = field(none, description='钩子函数')
    stream: bool = field(false, description='是否以流的形式响应')
    verify: bool = field(false, description='是否验证 ssl 证书')
    cert: union[str, tuple] = field(none, description='客户端 ssl 证书')

    class config:
        arbitrary_types_allowed = true

class responsemodel(basemodel):
    status: bool = field(true, description="响应状态")
    message: str = field('请求成功', description="响应信息")
    error: str = field('', description="错误信息")
    response: requests.response = field(none, description="响应对象")
    result: any = field({}, description="响应数据")

    class config:
        arbitrary_types_allowed = true

# coding: utf-8
import requests
from pyqt5.qtcore import pyqtsignal
from pydantic import basemodel
from queue import queue
from ..models import responsemodel, requestmodel

requests.packages.urllib3.disable_warnings()


class requestthread(qthread):
    responsechanged = pyqtsignal(basemodel)

    def __init__(self, parent=none):
        super().__init__(parent)
        self.queue = queue()

    def run(self):
        while not self.queue.empty():
            method = self.queue.get()  # type: requestmodel

            try:
                data = method.model_dump(mode='python')
                data['json'] = data.pop('json_', none)
                response = requests.request(**data)
                response.raise_for_status()
                self.responsechanged.emit(responsemodel(response=response))
            except requests.exceptions.requestexception as e:
                self.responsechanged.emit(responsemodel(status=false, message='请求失败',error=str(e)))

    def setrequest(self, method: requestmodel):
        self.queue.put(method)
        if not self.isrunning():
            self.start()

from pyqt5.qtwidgets import qapplication

from common import requestthread, requestmodel, responsemodel


def response_handler(response_model: responsemodel):
    if response_model.status:
        # todo: 当请求正确时处理逻辑
        print(response_model.response.text)
        pass
    else:
        # todo: 当请求错误时处理逻辑
        print(response_model.message, response_model.error)


# create a qapplication instance
app = qapplication([])

# create a request thread and start it
request_thread = requestthread()
request_thread.responsechanged.connect(response_handler)
request_thread.setrequest(requestmodel(url='http://www.baidu.com'))
request_thread.finished.connect(app.quit)
app.exec_()