08 Flask源码剖析之flask拓展点
1. 信号(源码)
信号,是在flask框架中为我们预留的钩子,让我们可以进行一些自定义操作。
pip3 install blinker
2. 根据flask项目的请求流程来进行设置扩展点
中间件
# 代码示例 from flask import Flask,render_template app = Flask(__name__) @app.route('/index')
def index():
return render_template('index.html') @app.route('/order')
def order():
return render_template('order.html') class MyMiddleware(object):
def __init__(self,old_app):
self.wsgi_app = old_app.wsgi_app def __call__(self, *args, **kwargs):
print('123')
result = self.wsgi_app(*args, **kwargs)
print('456')
return result app.wsgi_app = MyMiddleware(app) if __name__ == '__main__':
app.run()
当app_ctx被push到local中栈之后,会触发appcontext_pushed信号,之前注册在这个信号中的方法,就会被执行。
# 代码示例 from flask import Flask,render_template
from flask import signals app = Flask(__name__) @signals.appcontext_pushed.connect
def f1(arg):
print('appcontext_pushed信号f1被触发',arg) @signals.appcontext_pushed.connect
def f2(arg):
print('appcontext_pushed信号f2被触发',arg) @app.route('/index')
def index():
return render_template('index.html') @app.route('/order')
def order():
return render_template('order.html') if __name__ == '__main__':
app.run()
# app.__call__执行before_first_request扩展
# 代码示例 from flask import Flask,render_template app = Flask(__name__) @app.before_first_request
def f2():
print('before_first_requestf2被触发') @app.route('/index')
def index():
return render_template('index.html') @app.route('/order')
def order():
return render_template('order.html') if __name__ == '__main__':
app.run()
request_started信号
# 代码示例 from flask import Flask,render_template
from flask import signals
app = Flask(__name__) @signals.request_started.connect
def f3(arg):
print('request_started信号被触发',arg) @app.route('/index')
def index():
return render_template('index.html') @app.route('/order')
def order():
return render_template('order.html') if __name__ == '__main__':
app.run()
url_value_processor
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @app.url_value_preprocessor
def f5(endpoint,args):
print('f5') @app.route('/index/')
def index():
print('index')
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
before_reuqest
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @app.before_request
def f6():
g.xx = 123
print('f6') @app.route('/index/')
def index():
print('index')
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
视图函数
before_render_template / rendered_template
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @signals.before_render_template.connect
def f7(app, template, context):
print('f7') @signals.template_rendered.connect
def f8(app, template, context):
print('f8') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()after_request
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @app.after_request
def f9(response):
print('f9')
return response @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
request_finished
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @signals.request_finished.connect
def f10(app,response):
print('f10') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
got_request_exception
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @app.before_first_request
def test():
int('asdf') @signals.got_request_exception.connect
def f11(app,exception):
print('f11') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
teardown_request
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @app.teardown_request
def f12(exc):
print('f12') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
request_tearing_down
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @signals.request_tearing_down.connect
def f13(app,exc):
print('f13') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()
appcontext_popped
# 代码示例 from flask import Flask,render_template,g
from flask import signals
app = Flask(__name__) @signals.appcontext_popped.connect
def f14(app):
print('f14') @app.route('/index/')
def index():
return render_template('index.html') @app.route('/order')
def order():
print('order')
return render_template('order.html') if __name__ == '__main__':
app.run()扩展:flash
flash存值之后只能取一次
# 代码示例 from flask import Flask,render_template,flash,get_flashed_messages,session
from flask import signals
app = Flask(__name__)
app.secret_key = 'iuknsoiuwknlskjdf' @app.route('/index/')
def index():
# flash('123')
session['k1'] = 123
return render_template('index.html') @app.route('/order')
def order():
# messages = get_flashed_messages()
# print(messages)
val = session['k1']
del session['k1']
print(val)
return render_template('order.html') if __name__ == '__main__':
app.run()
总结:
- 关于flask内部共有14+个扩展点用于我们对flask框架内部进行定制,其中有:9个是信号。
# 源码示例
template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")