python泛型的实际场景使用之api接口回掉解析

最近做QT，也涉及到了接口的调用，那我们不能每个接口都去单独写一套数据的解析逻辑，so我按照我的场景抽象了请求和解析数据的逻辑代码，抛砖引玉。

首先我列出我的接口返回的2种数据格式如下：

第一种列表格式：

{
errcode: 0,
data: [
  {
    name: "71右2",
    age: 0,
    gender: 0,
    comment: "",
    measureTime: "2024-08-22T08:30:07.587+00:00",
    fileUrls: "["http:\/\/xxxxx\/mintti-audio\/tradition-data\/2024\/08\/22\/1724315411604063.wav"]",
    ctPicUrls: null,
    patientInfoPicUrls: null,
    ecgPicUrls: null
  },
  {
    name: "71右2",
    age: 0,
    gender: 0,
    comment: "",
    measureTime: "2024-08-22T08:30:07.587+00:00",
    fileUrls: "["http:\/\/xxxxx\/mintti-audio\/tradition-data\/2024\/08\/22\/1724315411604063.wav"]",
    ctPicUrls: null,
    patientInfoPicUrls: null,
    ecgPicUrls: null
  },
  ......  
],
errmsg: "成功"
}

第二种对象格式

{
  "errcode": 0,
  "data": {
    "url": "http://****:8088/aimed/app/storage/newFile/bat656psijl81ys69jed.wav"
  },
  "errmsg": "成功"
}

首先我们定义个基类，里面包含了，解析对象中每个字段的一个方法，

baseEntity.py文件

# 基础实体类
from typing import TypeVar, Type, Dict, Any

T = TypeVar('T', bound='BaseEntity')


class BaseEntity:
    def __init__(self):
        pass

    @classmethod  # 类方法
    def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
        """从字典动态解析类实例"""
        obj = cls()
        for key, value in data.items():
            # 设置属性
            setattr(obj, key, value)
        return obj

这里就涉及到泛型的概念，python叫协变，TypeVar就是定义个泛型类型，约束对象为BaseEntity。

新建一个列表数据格式的数据实体对象，measureDataModel.py

# 测量数据模型
from dataclasses import dataclass

from com.label.www.model.baseEntity import BaseEntity

@dataclass
class MeasureDataModel(BaseEntity):
    name:str
    age:int
    gender:int
    comment:str
    measureTime:str
    fileUrls:str
    ctPicUrls:str
    patientInfoPicUrls:str
    ecgPicUrls:str
    localFileUrls:str
    
    def __init__(self, name='', age=0, gender=0, comment='', measureTime='', fileUrls='',ctPicUrls='',patientInfoPicUrls='',ecgPicUrls='', localFileUrls=''):
        self.name = name
        self.age = age
        self.gender = gender  # 0:男 1:女
        self.comment = comment
        self.measureTime = measureTime
        self.fileUrls = fileUrls  # 源数据地址
        self.ctPicUrls = ctPicUrls
        self.patientInfoPicUrls = patientInfoPicUrls
        self.ecgPicUrls = ecgPicUrls
        self.localFileUrls = localFileUrls  # 缓存到本地数据
    def __str__(self):
        #输出各个字段的值
        return f"name:{self.name},age:{self.age},gender:{self.gender}," \
               f"comment:{self.comment},measureTime:{self.measureTime}," \
               f"fileUrls:{self.fileUrls},ctPicUrls:{self.ctPicUrls}," \
               f"patientInfoPicUrls:{self.patientInfoPicUrls},ecgPicUrls:{self.ecgPicUrls}," \
               f"localFileUrls:{self.localFileUrls}"

新建一个上传文件返回的 字典数据格式的对象 uploadFileEntity.py

from dataclasses import dataclass

from com.label.www.model.baseEntity import BaseEntity


# 上传数据返回
@dataclass
class UploadFileEntity(BaseEntity):
    url: str

    def __init__(self, url=''):
        self.url = url

    def __str__(self):
        # 输出各个字段的值
        return f"url:{self.url}"

新建返回数据reponse 基础的解析类，responseResultEntity.py

# 返回数据解析实体
from typing import Dict, Any, Type, Generic, TypeVar, Optional

from com.label.www.model.baseEntity import BaseEntity, T


# 定义类型变量
# response 基础的解析
class ResponseResultEntity(BaseEntity):

    def __init__(self):
        super().__init__()
        self.errcode = 0
        self.errmsg = ''
        self.data = Optional[Any]

    @classmethod  # 类方法
    def from_dict_base(cls, data: Dict[str, Any], clsChild: Type[T]) -> T:
        obj = super().from_dict(data)
        #data为列表
        if isinstance(data.get('data'), list):
            obj.data = []
            for i, item in enumerate(data['data']):
                _item = clsChild.from_dict(item)
                obj.data.append(_item)
        #data为字典
        elif isinstance(data.get('data'), dict):
            obj.data = {}
            obj.data = clsChild.from_dict(data['data'])
        #data为其他类型
        else:
            obj.data = data['data']
        return obj

    def __repr__(self):
        return 'ResponseResultEntity(errcode={}, errmsg={}, data={})'.format(self.errcode, self.errmsg, self.data)

基础类创建好后，我们新建一个发起请求的工具类，并单利模式实例化，代码如下，requestUtils.py

import json
import mimetypes
import os
from typing import Type

import requests

from com.label.www.model.baseEntity import BaseEntity, T
from com.label.www.model.measureDataModel import MeasureDataModel
from com.label.www.model.responseResultEntity import ResponseResultEntity
from com.label.www.model.uploadFileEntity import UploadFileEntity
from com.label.www.network.api import host, dataList, file_upload

# 定义 headers
headers = {
    # 'User-Agent': 'MyApp/1.0',
    'Accept-Encoding': 'gzip,deflate,br',
    'Authorization': 'Bearer your_token_here',
    'Content-Type': 'application/json',
}

#统一错误返回
def _error(msg='系统错误', errcode=500):
    responseResultEntity = ResponseResultEntity()
    responseResultEntity.errcode = errcode
    responseResultEntity.errmsg = msg
    responseResultEntity.data = []
    return responseResultEntity


# 请求类url配置
class RequestUtils:
    _instance = None
    _host = ""

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(RequestUtils, cls).__new__(cls)
            cls._host = host
        return cls._instance

    @staticmethod
    def getHeaders():
        # 设置token
        headers['Authorization'] = 'Bearer *******'
        return headers

    @staticmethod
    def get(url, cls: Type[T], params=None) -> ResponseResultEntity:
        url = f"{RequestUtils._host}{url}"
        print(f"请求开始: {url}")
        try:
            response = requests.get(url, params=params, headers=RequestUtils.getHeaders())
            print(f"请求结束: {response.url}")
            if response.status_code != 200:
                return _error(msg=f"请求失败: {response.status_code},{response.text}", errcode=response.status_code)
            # response.raise_for_status()  # 检查请求是否成功
            content = response.content.decode('utf-8-sig')
            # data = response.json()  # 直接使用 response.json() 也可以
            data = json.loads(content)  # 这样确保处理了 BOM
            print(f"data: {data}", type(data))
            return ResponseResultEntity.from_dict_base(data, cls)  # 动态解析为 ResponseEntity 实体
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return _error(msg=str(e))

    @staticmethod
    def post(url, cls: Type[T], params=None, data=None, pfiles=None) -> ResponseResultEntity:
        url = f"{RequestUtils._host}{url}"
        try:
            # timeout 请求超时时间
            response = requests.post(url,
                                     params=params,
                                     json=data,
                                     files=pfiles
                                     # headers=RequestUtils.getHeaders()
                                     )
            print(f"请求url: {response.url}")
            if response.status_code != 200:
                return _error(msg=f"请求失败: {response.status_code},{response.text}", errcode=response.status_code)
            # response.raise_for_status()  # 检查请求是否成功
            content = response.content.decode('utf-8-sig')
            # data = response.json()  # 直接使用 response.json() 也可以
            data = json.loads(content)  # 这样确保处理了 BOM
            print(f"data: {data}", type(data))
            return ResponseResultEntity.from_dict_base(data, cls)  # 动态解析为 ResponseEntity 实体
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return _error(msg=str(e))


if __name__ == '__main__':
    #get请求
    requestUtils = RequestUtils()
    res: ResponseResultEntity = requestUtils.get(dataList, MeasureDataModel, {'source': 2, 'page': 1, 'limit': 1000})
    print(res)

    print("#############")


    # 定义要上传的多个文件
    path = '/Users/*****/Documents/LabelSystem/local-import/2024/10/15/2022-07-06_16-36-42.wav'
    with open(path, 'rb') as f:
        filename = os.path.basename(path)
        # 获取 MIME 类型
        mime_type, _ = mimetypes.guess_type(path)
        print(mime_type)
        files = {'file': (filename, f, mime_type)}
        res2: ResponseResultEntity = requestUtils.post(file_upload, UploadFileEntity, pfiles=files)
        print(res2)

    pass

上传文件中的 host, 接口地址我定义到我的api.py 文件中，结构如下：

#api接口配置

host = "http://192.168.1.55:8088"

#获取数据列表
dataList = "/aimed/client/list"
#文件上传
file_upload = "/aimed/storage/upload/newFile"
#保存标注信息
saveAnnotationDate = "/aimed/annotation/saveAnnotationDate"
#根据标注id查询标注记录
getAnnotationData = "/aimed/annotation/getAnnotationData"

ok，这就是整个抽象的网络请求框架，接下来运行main函数，看下效果：

在请求的时候根据不同接口传递相应的对象 MeasureDataModel 和 UploadFileEntity，解析到相应的数据，是不是这样会简写很多代码。好啦！就这样吧，可以根据自己需求更改相关字段或结构。

作者blog：blog.codeceo.net