如何优雅的用PyQt访问http(仅供参考)
使用
pydantic
或dataclaass
创建一个数据存储对象
第一种
# 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()
第二种
model
模型
# 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_()