Flask自定义序列化超详细讲解
京金 人气:01、问题溯源
重点就是一个Flask.make_response,这里会做请求的响应的处理。
里面行代码:
elif isinstance(rv, dict): rv = jsonify(rv)
rv是我需要返回的响应:
{ 'msg': [ {'roleName': 'guest', 'access_list':[<accessName root>]}, {'roleName': 'admin', 'access_list': [...]} ], 'error_no': 0 }
rv是一个字典,但是msg的里面有部分东西无法序列化,jsonify里面返回如下。
return current_app.response_class( f"{dumps(data, indent=indent, separators=separators)}\n", mimetype=current_app.config["JSONIFY_MIMETYPE"], )
里面的data就是上面交道的类似json的数据(不是json,实际是对象)。
接下来:调用了flask下的自带的一个json库。
代码片段: M-1
_dump_arg_defaults(kwargs, app=app) return _json.dumps(obj, **kwargs)
代码片段: M -2
if cls is None: cls = JSONEncoder return cls( skipkeys=skipkeys, ensure_ascii=ensure_ascii, check_circular=check_circular, allow_nan=allow_nan, indent=indent, separators=separators, default=default, sort_keys=sort_keys, **kw).encode(obj)
这里的obj的:
{'msg': [{...}, {...}], 'error_no': 0}
就是我们之前讲到的东西。
代码片段: M-4,这个是内置的json库的default方法。内置的没有实现序列化,所以需要自己在某个步骤接入到这个序列化的过程。
def default(self, o): """Implement this method in a subclass such that it returns a serializable object for ``o``, or calls the base implementation (to raise a ``TypeError``). For example, to support arbitrary iterators, you could implement default like this:: def default(self, o): try: iterable = iter(o) except TypeError: pass else: return list(iterable) # Let the base class default method raise the TypeError return JSONEncoder.default(self, o) """ raise TypeError(f'Object of type {o.__class__.__name__} ' f'is not JSON serializable')
o在这里就是AccessOrm是实例对象,所以他报错说这个:
Object of type AccessOrm is not JSON serializable
归因:就是flask没有自带实现对类的序列化
解决: 就是通过flask的机制,绑定一个序列化类。
2、flask序列化
flask代码里面写了:
class JSONEncoder(_json.JSONEncoder): """The default JSON encoder. Handles extra types compared to the built-in :class:`json.JSONEncoder`. - :class:`datetime.datetime` and :class:`datetime.date` are serialized to :rfc:`822` strings. This is the same as the HTTP date format. - :class:`uuid.UUID` is serialized to a string. - :class:`dataclasses.dataclass` is passed to :func:`dataclasses.asdict`. - :class:`~markupsafe.Markup` (or any object with a ``__html__`` method) will call the ``__html__`` method to get a string. Assign a subclass of this to :attr:`flask.Flask.json_encoder` or :attr:`flask.Blueprint.json_encoder` to override the default. """
在片段代码M-1中:
_dump_arg_defaults(kwargs, app=app)
函数内容如下:
def _dump_arg_defaults( kwargs: t.Dict[str, t.Any], app: t.Optional["Flask"] = None ) -> None: """Inject default arguments for dump functions.""" if app is None: app = current_app if app: cls = app.json_encoder #app的json_encoder bp = app.blueprints.get(request.blueprint) if request else None # type: ignore if bp is not None and bp.json_encoder is not None: cls = bp.json_encoder #这里设置蓝图的json_encoder,蓝图的优先级高于app.json_encoder # Only set a custom encoder if it has custom behavior. This is # faster on PyPy. if cls is not _json.JSONEncoder: kwargs.setdefault("cls", cls) kwargs.setdefault("cls", cls) kwargs.setdefault("ensure_ascii", app.config["JSON_AS_ASCII"]) kwargs.setdefault("sort_keys", app.config["JSON_SORT_KEYS"]) else: kwargs.setdefault("sort_keys", True) kwargs.setdefault("cls", JSONEncoder)
3、解决方法
使用自带的序列化类
在入口函数setup.py中写入
from flask import Flask from flask.json import JSONEncoder from config import Config,LogConfig from db import init_db from scripts import user_cli from flask_bcrypt import Bcrypt from utils import RedisPool class ExtendJSONEncoder(JSONEncoder): def default(self, o): if getattr(o,'toJson'): return o.toJson(o) else: return super().default(o) flask_bcrypt = Bcrypt() def create_app(): Flask.json_encoder = ExtendJSONEncoder #之前在with app.app_context() #或者在app实例化之后,修改app的JSONEncoder 都没成功,这里简单粗暴一点,直接修改Flask的。 app = Flask(__name__) app.config.from_object(Config) LogConfig.openLog() from utils import initException,initBeforeRequestHandle with app.app_context(): init_db(app) RedisPool(app) initException(app) initBeforeRequestHandle(app) flask_bcrypt.init_app(app) app.cli.add_command(user_cli) import controller for bp in controller.__all__: app.register_blueprint(controller.__dict__[bp]) return app
这时候,我的orm类需要一个toJson方法。
class AccessOrm(Base): __tablename__ = 'access' id = Column(Integer, primary_key=True) accessName = Column(String(255), nullable=True) def __repr__(self) -> str: return "<accessName {}>".format( self.accessName, ) def toJson(self,o): return { 'accessName': o.accessName, }
加载全部内容