统一的 JSON 序列化器
全项目数据层的基石之一:数据库对象可以直接 return,由序列化器自动转成 JSON,同时在序列化前支持精确的「隐藏 / 追加」字段控制。
核心代码分布在两个文件:
app/core/json_encoder.py—— 自定义 Flask 的JSONEncoderapp/core/db.py——JSONSerializerMixin(给 Model 注入hide/append/keys能力)
解决的问题
大多数后端要返回数据时,得做两件麻烦事:
- 手动
.to_dict():每个 Model 写一个转换方法,字段一变就要同步维护,容易漏。 - 处理敏感字段:比如订单表的
prepay_id(微信支付预支付 ID)不该给前端,通常要在返回前del掉,散落在接口层各处。
这套设计用「Mixin + 自定义编码器」一次性解决两者。
一、Model 自带「隐藏 / 追加」能力(JSONSerializerMixin)
app/core/db.py 里的 JSONSerializerMixin,给所有 Model 混入了两个方法:
# 用法:在接口里对查到的对象操作
order = Order.query.get_or_404(id).hide('prepay_id') # 隐藏敏感字段
group = Group.get_or_404(id).append('auth_list') # 追加临时字段
# 分页查询也能整批操作
paged = Order.query.paginate(...).hide('prepay_id') # 每一条都隐藏
paged = Order.query.paginate(...).append('extra') # 每一条都追加关键:keys() / __getitem__() 让对象可被 dict() 处理
def keys(self):
'''返回序列化时包含哪些字段'''
return self.fields
def __getitem__(self, item):
attr = getattr(self, item)
# 字符串字段若本身是 JSON,自动解析成对象
if isinstance(attr, str):
try:
attr = json.loads(attr)
except ValueError:
pass
# 时间字段自动从时间戳转成可读格式
if item in ['create_time', 'update_time', 'delete_time']:
attr = strftime('%Y-%m-%d %H:%M:%S', localtime(attr))
return attr实现了 keys() 和 __getitem__,Model 实例就可以被 dict(obj) 转换(Python 的 dict 协议)——这是后续序列化器能直接返回对象的前提。
初始化时按表裁剪字段
@orm.reconstructor
def init_on_load(self):
self._locked = False
self._locked_fileds = []
self._exclude = []
self._set_fields() # 由子类设置 exclude
self.__prune_fields() # 实际字段 = 表全部列 - exclude
def __prune_fields(self):
all_columns = inspect(self.__class__).columns.keys()
self.fields = list(set(all_columns) - set(self._exclude))EntityModel默认把create_time/update_time/delete_time放进_exclude(不随dict输出,逻辑里再按需append)。- 每个 Model 可通过重写
_set_fields自定义默认排除字段——表字段天然不直接暴露,暴露是显式行为。
字段锁:lock_fileds 防止序列化阶段再改字段集合
def hide(self, *keys):
for key in keys:
if hasattr(self, key):
if not self._locked:
self._locked_fileds.append(key)
self.fields.remove(key) # 业务期隐藏
if self._locked and key not in self._locked_fileds:
self.fields.remove(key) # 序列化期也能隐藏「未被锁住的」字段lock_fileds() 在业务逻辑结束后(序列化前)调用,之后不能对业务期已操作过的字段再 hide/append——防止序列化阶段意外改动字段集合、或在多层装饰器间被误改。
二、自定义 JSON Encoder:直接返回对象
app/core/json_encoder.py:
class JSONEncoder(_JSONEncoder):
def default(self, obj):
# 若是数据库实例(具备 keys 协议)
if hasattr(obj, 'keys') and hasattr(obj, '__getitem__'):
obj.lock_fileds() # 序列化前锁定字段
return dict(obj)
# datetime / date 自动格式化
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
if isinstance(obj, date):
return obj.strftime('%Y-%m-%d')
raise ServerError()在 app/__init__.py 里 app.json_encoder = JSONEncoder 注册后,任何接口直接 return Success(order_obj) 或 return dict(order_obj) 都能正确序列化,订单对象自动走 hide/append 后的字段集合。
三、完整链路
接口层 order.hide('prepay_id') # 业务期:隐藏敏感字段
group.append('auth_list') # 业务期:注入临时字段
↓
return Success(obj) # 业务结束,交给响应
↓
JSONEncoder.default(obj) # 序列化前调 lock_fileds() 锁定
↓
dict(obj) → keys() → 序列化字段集合 # 按 fields 输出
↓
返回 { error_code, msg, data } # 统一格式 JSON真实用例
app/api/v1/order.py —— 隐藏支付预支付号:
@api.route('/<int:id>', methods=['GET'])
@auth.login_required
def get_order(id):
order = Order.query.get_or_404(id).hide('prepay_id')
return Success(order)app/api/cms/group.py —— 给权限组临时追加其权限列表:
@api.route('/<int:id>', methods=['GET'])
def get_group(id):
group = Group.get_or_404(id)
group.append('auth_list') # 序列化时自动带上 auth_list
return Success(group)设计价值
- 契约即代码:返回给前端哪些字段,在业务代码里通过
hide/append显式声明,而不是散落在.to_dict()。 - 安全默认:表字段不主动暴露;时间、JSON 字符串字段自动规整,接口层无需重复处理。
- 链式可读:
query.hide(...).paginate().hide(...)一步到位,也支持分页批量操作。 - 防误改:
lock_fileds锁住业务期字段集合,序列化阶段不会再被意外改动。
能带走的思想
先想清楚「数据以什么形态出境」,用一个统一的序列化器收口,而不是让每个 Model 各写一个 to_dict()、返回前再手动删字段。配合「默认不暴露、需要时显式 hide/append」的策略,敏感字段不易泄漏。任何语言的 API 都能借鉴:序列化独立成层,字段白名单集中控制。
下一步:参数校验层