flask signal
介绍
flask通过blinker实现信号机制。
blinker
blinker是python里 的信号库，可用来对组建解耦。一个例子
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
from blinker import signal
print_signal = signal(‘print’)
@print_signal.connect_via()
def printer(sender):
print sender
print_signal.send(‘obj’)
|
flask signal
flask core中的signal
flask/signals.py
1
2
3
4
5
6
7
8
9
10
11
12
13
| from blinker import Namespace
_signals = Namespace()
template_rendered = _signals.signal(‘template-rendered’)
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’)
|
signal send
以request_started为例。
先看flask wsgi app 的实现
flask/app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class Flask(_PackageBoundObject):
…
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
ctx.push()
error = None
try:
try:
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.make_response(self.handle_exception(e))
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
|
当调用self.full_dispatch_request()时
1
2
3
4
5
6
7
8
9
10
11
12
13
| def full_dispatch_request(self):
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
response = self.make_response(rv)
response = self.process_response(response)
request_finished.send(self, response=response)
return response
|
该方法会先发送一个signal，然后才进行request的预处理。
在flask中用装饰器的方式使用
1
2
3
| @request_started.connect_via(app)
def start(sender, **extra):
print sender.name
|
