详解Python的Flask框架中的signals信号机制

pip install blinker

from blinker import Namespace
product_saved = Namespace()

from flask.singles import Namespace

product_saved.send(app, product=product)

@product_saved.connect_via(app)
def updateCache(app, product):
  print(product)

def log_template_renders(sender, template, context, **extra):
  sender.logger.debug('Rendering template "%s" with context %s',
            template.name or 'string template',
            context)
from flask import template_rendered
template_rendered.connect(log_template_renders, app)

def log_request(sender, **extra):
  sender.logger.debug('Request context is set up')
from flask import request_started
request_started.connect(log_request, app)
flask.request_finished

def log_response(sender, response, **extra):
  sender.logger.debug('Request context is about to close down. '
            'Response: %s', response)
from flask import request_finished
request_finished.connect(log_response, app)
flask.got_request_exception

def log_exception(sender, exception, **extra):
  sender.logger.debug('Got exception during processing: %s', exception)
from flask import got_request_exception
got_request_exception.connect(log_exception, app)
flask.request_tearing_down

def close_db_connection(sender, **extra):
  session.close()from flask import appcontext_tearing_down
request_tearing_down.connect(close_db_connection, app)

def close_db_connection(sender, **extra):
  session.close()from flask import request_tearing_down
appcontext_tearing_down.connect(close_db_connection, app)

from contextlib import contextmanagerfrom
flask import appcontext_pushed
@contextmanagerdef user_set(app, user):
  def handler(sender, **kwargs):
    g.user = user
  with appcontext_pushed.connected_to(handler, app):
    yield

def test_user_me(self):
  with user_set(app, 'john'):
    c = app.test_client()
    resp = c.get('/users/me')
    assert resp.data == 'username=john'
New in version 0.10.

recorded = []def record(sender, message, category, **extra):
  recorded.append((message, category))
from flask import message_flashed
message_flashed.connect(record, app)