怎么做企业网站推广需要多少钱_重庆广告制作加工厂_互联网营销师培训机构_足球排行榜前十名

使用pydantic或dataclaass创建一个数据存储对象

第一种

# coding: utf-8
from typing import Anyimport requests
from pydantic import Field, BaseModelclass 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, Slotfrom common.models import ResponseModel
from loguru import loggerclass RequestThread(QThread):modelChanged = Signal(ResponseModel)def __init__(self, parent=None):super().__init__(parent)self.request = Nonedef run(self):if not self.request:returntry: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 = requestif not self.isRunning():self.start()

使用函数将请求对象加入至线程中

import requests
from PySide6.QtCore import QCoreApplicationfrom common.models import ResponseModel
from request_thread import RequestThreaddef baidu_request():response = requests.get('http://www.baidu.com')response.encoding = 'utf-8'return response.textdef response_handler(response_model: ResponseModel):if response_model.status:# TODO: 当请求正确时处理逻辑print(response_model.result)passelse:# 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, Anyfrom pydantic import BaseModel, Fieldclass 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 = Trueclass 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, RequestModelrequests.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: RequestModeltry: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 QApplicationfrom common import RequestThread, RequestModel, ResponseModeldef response_handler(response_model: ResponseModel):if response_model.status:# TODO: 当请求正确时处理逻辑print(response_model.response.text)passelse:# 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_()