python3--使用解析protocol反序列化 数据库里bl

使用了这个github库 解决了class无法生成问题
https://github.com/danielgtaylor/python-betterproto

message 文件

依赖

python3 -m pip install "betterproto[compiler]"

编译proto文件

mkdir lib
protoc -I . --python_betterproto_out=lib ucloud.proto
protoc -I . --python_betterproto_out=lib uhc.115000.120000.proto

反序列化

序列化的数据存到了blob字段里 UHostPmResource是事先定义好的message
直接就能解析出来了

#!/usr/bin/env python3

# 10.68.130.103
# uhc
# SELECT * FROM t_mixcloud_host_resource WHERE id='b080c7f0-2168-4809-aa8b-7c05afb92719' LIMIT 10;

#import encodings
#import imp
from typing import TypedDict
import pymysql
#import _pickle as cPickle
#from example import openrtb_pb2

from google.protobuf import json_format


class MysqlInfoDict(TypedDict):
    host: str
    user: str
    passwd: str
    db: str
    port: int
    charset: str

class MySqlUtils:
    def __init__(self, db_info: MysqlInfoDict):
        self.conn = pymysql.connect(
            host = db_info["host"],\
            user = db_info["user"],\
            passwd = db_info["passwd"],\
            db = db_info["db"],\
            port = db_info["port"],\
            charset = db_info["charset"]
        )
    def exec(self, sql):
        cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
        print("exec sql: %s", sql)
        try:
            cur.execute(sql)
            res = cur.fetchall()
        except Exception as e:
            print("exec sql failed ! %s" % str(e))
            raise()
        finally:
            cur.close()
        print("result is %s" % res)
        return res

# def pb_to_json(pbStringRequest):
#     """将pbstring转化为jsonStringResponse返回"""
#     msg = openrtb_pb2.param()
#     msg.ParseFromString(pbStringRequest)
#     jsonStringRequest=json_format.MessageToJson(msg)
#     return jsonStringRequest

def main():
    db_info = MysqlInfoDict(
        host = "192.168.100.100",
        user = "root",
        passwd = "123456",
        db = "test",
        port = 3306,
        charset = "utf8"
    )
    print(db_info["host"])
    mysql_cli = MySqlUtils(db_info)
    sql = "SELECT * FROM t_mixcloud_host_resource WHERE id='b080c7f0-2168-4809-aa8b-7c05afb92719' LIMIT 10;"
    res = mysql_cli.exec(sql)
    blob_val = res[0]["value"]

    #data =  cPickle.loads(_base255.decode(blob_val))
    #data = str(blob_val, encoding = "latin1")
    #data = blob_val.decode(encoding="ISO-8859-1", errors="strict")
    #data = cPickle.loads(blob_val, encoding="ISO-8859-1")

    from message import uhc
    data = uhc.UHostPmResource().parse(blob_val)
    print(type(blob_val), data)


if __name__ == "__main__":
    main()