使用 Flask 正确重载 json 编码和解码
Proper overloading of json encoding and decoding with Flask
我正在尝试向 Flask 添加一些重载 JSON encoder/decoder 以添加日期时间 encoding/decoding 但仅通过 'hack'.
成功
from flask import Flask, flash, url_for, redirect, render_template_string
from flask.json import JSONEncoder, JSONDecoder
template = """
<!DOCTYPE html>
<html><head><title>Test JSON encoder/decoder</title></head><body>
{% with messages = get_flashed_messages(with_categories=true) %}{% if messages %}{% for message in messages %}
<p>Flash: {{ message }}</p>
{% endfor %}{% endif %}{% endwith %}
<p>Flash should be: ['Flash message', 'success']</p>
<p><a href="{{ url_for('index') }}">Try again</a></p>
</body></html>
"""
class CustomJSONEncoder(JSONEncoder):
""" Do nothing custom json encoder """
def default(self, obj):
# My custom logic here
# ...
# or
return super(CustomJSONEncoder, self).defaults(obj)
class CustomJSONDecoder(JSONDecoder):
""" Do nothing custom json decoder """
def __init__(self, *args, **kargs):
_ = kargs.pop('object_hook', None)
super(CustomJSONDecoder, self).__init__(object_hook=self.decoder, *args, **kargs)
def decoder(self, d):
# My custom logic here
# ...
# or
return d
app = Flask(__name__, static_url_path='')
app.config['SECRET_KEY'] = 'secret-key'
app.json_encoder = CustomJSONEncoder
app.json_decoder = CustomJSONDecoder
@app.route('/')
def index():
flash('Flash message', 'success')
return redirect(url_for('display'))
@app.route('/b')
def display():
return render_template_string(template)
if __name__ == '__main__':
app.run(debug=True, port=5200)
技巧是我应该从 Flask.sessions.TaggedJSONSerializer 中复制一些代码,像这样:
import uuid
from base64 import b64decode
from werkzeug.http import parse_date
from markupsafe import Markup
from flask._compat import iteritems
class CustomJSONDecoder(JSONDecoder):
""" Do nothing custom json decoder """
def __init__(self, *args, **kargs):
_ = kargs.pop('object_hook', None)
super(CustomJSONDecoder, self).__init__(object_hook=self.decoder, *args, **kargs)
def decode(self, d):
# My custom logic here
# ...
# Copy of the code from Flask.sessions.TaggedJSONSerializer(object).loads(self, value).object_hook(obj)
if len(d) == 1:
the_key, the_value = next(iteritems(d))
if the_key == ' t':
return tuple(the_value)
elif the_key == ' u':
return uuid.UUID(the_value)
elif the_key == ' b':
return b64decode(the_value)
elif the_key == ' m':
return Markup(the_value)
elif the_key == ' d':
return parse_date(the_value)
return d
我会这样做吗 'correctly' 还是我错过了什么?
您可以通过显式调用它的 default() 方法来使用基础 class 的功能。我已经在我的自定义 JSONEncoder 中成功地做到了:
class CustomJSONEncoder(JSONEncoder):
def default(self, obj):
# Calling custom encode function:
jsonString = HelperFunctions.jsonEncodeHandler(obj)
if (jsonString != obj): # Encode function has done something
return jsonString # Return that
return JSONEncoder.default(self, obj) # else let the base class do the work
然而,在解码器中,您应该记住传递给 __init__()
函数的对象钩子并从您自己的钩子中调用它:
class CustomJSONDecoder(JSONDecoder):
def __init__(self, *args, **kwargs):
self.orig_obj_hook = kwargs.pop("object_hook", None)
super(CustomJSONDecoder, self).__init__(*args,
object_hook=self.custom_obj_hook, **kwargs)
def custom_obj_hook(self, dct):
# Calling custom decode function:
dct = HelperFunctions.jsonDecodeHandler(dct)
if (self.orig_obj_hook): # Do we have another hook to call?
return self.orig_obj_hook(dct) # Yes: then do it
return dct # No: just return the decoded dict
顺便说一句:你的解码器有一个错字:你在基 class 中注册的对象钩子被命名为 self.decoder
,但成员被定义为 def decode(...)
(没有r 最后)。在您的示例中,您注册了一个空钩子并且永远不会调用 decode()
。
请注意,您必须告诉您的 Flask 应用程序它将使用什么编码器:
app.json_encoder = CustomJSONEncoder
这解决了我的问题。
我正在尝试向 Flask 添加一些重载 JSON encoder/decoder 以添加日期时间 encoding/decoding 但仅通过 'hack'.
成功from flask import Flask, flash, url_for, redirect, render_template_string
from flask.json import JSONEncoder, JSONDecoder
template = """
<!DOCTYPE html>
<html><head><title>Test JSON encoder/decoder</title></head><body>
{% with messages = get_flashed_messages(with_categories=true) %}{% if messages %}{% for message in messages %}
<p>Flash: {{ message }}</p>
{% endfor %}{% endif %}{% endwith %}
<p>Flash should be: ['Flash message', 'success']</p>
<p><a href="{{ url_for('index') }}">Try again</a></p>
</body></html>
"""
class CustomJSONEncoder(JSONEncoder):
""" Do nothing custom json encoder """
def default(self, obj):
# My custom logic here
# ...
# or
return super(CustomJSONEncoder, self).defaults(obj)
class CustomJSONDecoder(JSONDecoder):
""" Do nothing custom json decoder """
def __init__(self, *args, **kargs):
_ = kargs.pop('object_hook', None)
super(CustomJSONDecoder, self).__init__(object_hook=self.decoder, *args, **kargs)
def decoder(self, d):
# My custom logic here
# ...
# or
return d
app = Flask(__name__, static_url_path='')
app.config['SECRET_KEY'] = 'secret-key'
app.json_encoder = CustomJSONEncoder
app.json_decoder = CustomJSONDecoder
@app.route('/')
def index():
flash('Flash message', 'success')
return redirect(url_for('display'))
@app.route('/b')
def display():
return render_template_string(template)
if __name__ == '__main__':
app.run(debug=True, port=5200)
技巧是我应该从 Flask.sessions.TaggedJSONSerializer 中复制一些代码,像这样:
import uuid
from base64 import b64decode
from werkzeug.http import parse_date
from markupsafe import Markup
from flask._compat import iteritems
class CustomJSONDecoder(JSONDecoder):
""" Do nothing custom json decoder """
def __init__(self, *args, **kargs):
_ = kargs.pop('object_hook', None)
super(CustomJSONDecoder, self).__init__(object_hook=self.decoder, *args, **kargs)
def decode(self, d):
# My custom logic here
# ...
# Copy of the code from Flask.sessions.TaggedJSONSerializer(object).loads(self, value).object_hook(obj)
if len(d) == 1:
the_key, the_value = next(iteritems(d))
if the_key == ' t':
return tuple(the_value)
elif the_key == ' u':
return uuid.UUID(the_value)
elif the_key == ' b':
return b64decode(the_value)
elif the_key == ' m':
return Markup(the_value)
elif the_key == ' d':
return parse_date(the_value)
return d
我会这样做吗 'correctly' 还是我错过了什么?
您可以通过显式调用它的 default() 方法来使用基础 class 的功能。我已经在我的自定义 JSONEncoder 中成功地做到了:
class CustomJSONEncoder(JSONEncoder):
def default(self, obj):
# Calling custom encode function:
jsonString = HelperFunctions.jsonEncodeHandler(obj)
if (jsonString != obj): # Encode function has done something
return jsonString # Return that
return JSONEncoder.default(self, obj) # else let the base class do the work
然而,在解码器中,您应该记住传递给 __init__()
函数的对象钩子并从您自己的钩子中调用它:
class CustomJSONDecoder(JSONDecoder):
def __init__(self, *args, **kwargs):
self.orig_obj_hook = kwargs.pop("object_hook", None)
super(CustomJSONDecoder, self).__init__(*args,
object_hook=self.custom_obj_hook, **kwargs)
def custom_obj_hook(self, dct):
# Calling custom decode function:
dct = HelperFunctions.jsonDecodeHandler(dct)
if (self.orig_obj_hook): # Do we have another hook to call?
return self.orig_obj_hook(dct) # Yes: then do it
return dct # No: just return the decoded dict
顺便说一句:你的解码器有一个错字:你在基 class 中注册的对象钩子被命名为 self.decoder
,但成员被定义为 def decode(...)
(没有r 最后)。在您的示例中,您注册了一个空钩子并且永远不会调用 decode()
。
请注意,您必须告诉您的 Flask 应用程序它将使用什么编码器:
app.json_encoder = CustomJSONEncoder
这解决了我的问题。