快捷導(dǎo)航

詳解如何優(yōu)雅的用PyQt訪(fǎng)問(wèn)http

更新時(shí)間：2024年11月29日 09:20:58 作者：赤鳶QAQ

這篇文章主要我打開(kāi)詳細(xì)介紹了如何優(yōu)雅的用PyQt實(shí)現(xiàn)訪(fǎng)問(wèn)http,文中的示例代碼講解詳細(xì),具有一定的借鑒價(jià)值,感興趣的小伙伴可以了解下

使用pydantic或dataclaass創(chuàng)建一個(gè)數(shù)據(jù)存儲(chǔ)對(duì)象

第一種

# coding: utf-8
from typing import Any

import requests
from pydantic import Field, BaseModel


class ResponseModel(BaseModel):
    status: bool = Field(True, description="響應(yīng)狀態(tài)")
    message: str = Field('請(qǐng)求成功', description="響應(yīng)信息")
    error: str = Field('', description="錯(cuò)誤信息")
    response: requests.Response = Field(None, description="響應(yīng)對(duì)象")
    result: Any = Field({}, description="響應(yīng)數(shù)據(jù)")

    class Config:
        arbitrary_types_allowed = True

線(xiàn)程

# 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='請(qǐng)求失敗',
                    error=e,
                )
            )

    def setRequest(self, request):
        self.request = request
        if not self.isRunning():
            self.start()

使用函數(shù)將請(qǐng)求對(duì)象加入至線(xiàn)程中

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: 當(dāng)請(qǐng)求正確時(shí)處理邏輯
        print(response_model.result)
        pass
    else:
        # TODO: 當(dāng)請(qǐng)求錯(cuò)誤時(shí)處理邏輯
        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()

第二種

model模型

# coding: utf-8
from typing import Union, Any

from pydantic import BaseModel, Field


class RequestModel(BaseModel):
    method: str = Field('GET', description='請(qǐng)求方法，如 GET、POST、PUT、DELETE')
    url: str = Field(..., description='請(qǐng)求的 URL 地址')
    params: dict = Field(None, description='請(qǐng)求參數(shù)，如 GET 請(qǐng)求時(shí)附帶的參數(shù)')
    data: dict = Field(None, description='請(qǐng)求數(shù)據(jù)，如 POST 請(qǐng)求時(shí)提交的數(shù)據(jù)')
    json_: dict = Field(None, description='請(qǐng)求數(shù)據(jù)，如 POST 請(qǐng)求時(shí)提交的 json 數(shù)據(jù)', alias='json')
    headers: dict = Field(None, description='請(qǐng)求頭，如 Content-Type、User-Agent 等')
    cookies: dict = Field(None, description='請(qǐng)求 cookies，如登錄后獲取的 cookie')
    files: dict = Field(None, description='上傳的文件，如 POST 請(qǐng)求時(shí)上傳的文件')
    auth: Union[tuple, list] = Field(None, description='HTTP 認(rèn)證，如 Basic 認(rèn)證')
    timeout: int = Field(None, description='請(qǐng)求超時(shí)時(shí)間，單位為秒')
    allow_redirects: bool = Field(True, description='是否允許重定向')
    proxies: dict = Field(None, description='代理設(shè)置')
    hooks: Any = Field(None, description='鉤子函數(shù)')
    stream: bool = Field(False, description='是否以流的形式響應(yīng)')
    verify: bool = Field(False, description='是否驗(yàn)證 SSL 證書(shū)')
    cert: Union[str, tuple] = Field(None, description='客戶(hù)端 SSL 證書(shū)')

    class Config:
        arbitrary_types_allowed = True

class ResponseModel(BaseModel):
    status: bool = Field(True, description="響應(yīng)狀態(tài)")
    message: str = Field('請(qǐng)求成功', description="響應(yīng)信息")
    error: str = Field('', description="錯(cuò)誤信息")
    response: requests.Response = Field(None, description="響應(yīng)對(duì)象")
    result: Any = Field({}, description="響應(yīng)數(shù)據(jù)")

    class Config:
        arbitrary_types_allowed = True

請(qǐng)求

# 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='請(qǐng)求失敗',error=str(e)))

    def setRequest(self, method: RequestModel):
        self.queue.put(method)
        if not self.isRunning():
            self.start()

運(yùn)用

from PyQt5.QtWidgets import QApplication

from common import RequestThread, RequestModel, ResponseModel


def response_handler(response_model: ResponseModel):
    if response_model.status:
        # TODO: 當(dāng)請(qǐng)求正確時(shí)處理邏輯
        print(response_model.response.text)
        pass
    else:
        # TODO: 當(dāng)請(qǐng)求錯(cuò)誤時(shí)處理邏輯
        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_()

到此這篇關(guān)于詳解如何優(yōu)雅的用PyQt訪(fǎng)問(wèn)http的文章就介紹到這了,更多相關(guān)PyQt訪(fǎng)問(wèn)http內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: