个人总结
command.py
file.py
time.py
excel.py
with 的原理和用法
subprocess 模块
cline 提示词
PathLike
pathlib
peewee
生成随机ID
FastApi 使用 peewee
http connect
Dataframe Protocol
pyarrow
overload 函数重载
1111
peewee 线程安全
Python 日志模块
rabbitmq 问答
多进程 daemon 参数
记录异常
dict.py
多进程
忽略异常
timer.py
Ctrl+C 退出while
发布pip包
Bolo组件
迭代器与生成器
dataclass
单例模式(二)
单例模式(一)
errors.py
enum.py
每日学习
本文档使用 MrDoc 发布
-
+
首页
Bolo组件
### 概述 `Bolo`是一个python类,它可以用来在不同的系统之间传递对象。 示例: ``` data = [1,[2,3], datetime.now()] r = Bolo.encode(data) print(r.json()) # 可以将r.json()的结果发送给其它服务 rr = Bolo.from_json(r.json()) # 通过decode()可以获得原始数据,类型也都是正确的 print(rr.decode()) print(r.json() == rr.json()) ``` 支持的类型: - `int/float/str` - `list/tuple/dict` - `date/datetime` - `pandas.DataFrame` ### 用途 在不同的python服务之间传递数据,其实可以使用`pickle`,但是这不利于日志的记录,因为`pickle`产生的对象是不可阅读的,日志记录下来也没法阅读。 因此`Bolo`可以替代`pickle`,用于不同服务之间的数据传递,便于日志记录。 ### 待解决的问题 1. 数据膨胀的厉害,json原生支持的数据其实不需要再封装了 ### 源码 ``` import pandas from datetime import date, datetime class Bolo(object): def __init__(self, data, _type, _datetime_format="%Y-%m-%d %H:%M:%S", _date_format="%Y-%m-%d"): if isinstance(data, list): data2 = [] for d in data: if isinstance(d, dict) and "_type" in d: data2.append(Bolo(**d)) else: data2.append(d) elif isinstance(data, dict) and "_type" in data: data2 = Bolo(**data) elif isinstance(data, dict): data2 = {} for key, val in data.items(): data2[key] = Bolo(**val) elif isinstance(data, (str, int, float)): data2 = data else: raise TypeError(f"invalid data with type={type(data)}") self.data = data2 self._type = _type self._date_format = _date_format self._datetime_format = _datetime_format def decode(self): result = None if self._type == "pandas.core.frame.DataFrame": result = pandas.read_json(self.data) elif self._type == "datetime.datetime": result = datetime.strptime(self.data, self._datetime_format) elif self._type == "datetime.date": result = datetime.strptime(self.data, self._date_format) elif isinstance(self.data, list): result = [d.decode() if isinstance(d, Bolo) else d for d in self.data] elif isinstance(self.data, dict): result = {} for key, val in self.data.items(): result[key] = val.decode() if isinstance(val, Bolo) else val else: result = self.data return result @staticmethod def encode(obj, _datetime_format="%Y-%m-%d %H:%M:%S", _date_format="%Y-%m-%d"): data, _type = None, None if isinstance(obj, (int, float, str)): data = obj elif isinstance(obj, (list, tuple)): data = [Bolo.encode(item, _datetime_format=_datetime_format, _date_format=_date_format) for item in obj] elif isinstance(obj, dict): data = {} for key, val in obj.items(): data[key] = Bolo.encode(val, _datetime_format=_datetime_format, _date_format=_date_format) elif isinstance(obj, datetime): data = obj.strftime(_datetime_format) _type = "datetime.datetime" elif isinstance(obj, date): data = obj.strftime(_date_format) _type = "datetime.date" elif isinstance(obj, pandas.DataFrame): data = obj.to_json() _type = "pandas.core.frame.DataFrame" return Bolo(data=data, _type=_type) def json(self): r = None if isinstance(self.data, list): r = [d.json() if isinstance(d, Bolo) else d for d in self.data] elif isinstance(self.data, dict): r = {} for key, val in self.data.items(): r[key] = val.json() if isinstance(val, Bolo) else val else: r = self.data result = { "data": r, "_type": self._type, "_date_format": self._date_format, "_datetime_format": self._datetime_format, } return result @staticmethod def from_json(data: dict): if "_type" not in data: raise ValueError("invalid data") return Bolo(**data) if __name__ == "__main__": data0 = 1 data1 = [1,2,3] data2 = [1,[2,3]] data3 = [1,[2,3], datetime.now()] data4 = {"name":"gaojian"} data5 = pandas.DataFrame([[1, 1, 2], [2, 3,4], [3, 5, 6]], columns=["id", "name", "age"]) # r0 = Bolo.encode(data0) # r1 = Bolo.encode(data1) # r2 = Bolo.encode(data2) # r3 = Bolo.encode(data3) # r4 = Bolo.encode(data4) r5 = Bolo.encode(data5) # print(r0, r0.data, r0.decode()) # print(r1, r1.data) # print(r1.decode()) # print(r2.decode()) # print(r3.decode()) # print(r4.decode()) # print(r5.decode()) # print(r1.json()) # print(r2.json()) # print(r3.json()) # print(r4.json()) print(r5.json()) # rr1 = Bolo.from_json(r1.json()) # print(rr1.json()) # rr3 = Bolo.from_json(r3.json()) # print(rr3.json()) # print(rr3.decode()) rr5 = Bolo.from_json(r5.json()) print(rr5.json()) print(r5.json() == rr5.json()) print(rr5.decode()) ```
gaojian
2023年2月24日 15:43
分享文档
收藏文档
上一篇
下一篇
微信扫一扫
复制链接
手机扫一扫进行分享
复制链接
关于 MrDoc
觅思文档MrDoc
是
州的先生
开发并开源的在线文档系统,其适合作为个人和小型团队的云笔记、文档和知识库管理工具。
如果觅思文档给你或你的团队带来了帮助,欢迎对作者进行一些打赏捐助,这将有力支持作者持续投入精力更新和维护觅思文档,感谢你的捐助!
>>>捐助鸣谢列表
微信
支付宝
QQ
PayPal
Markdown文件
分享
链接
类型
密码
更新密码