Catch signals in Flask Blueprint

Question

I'm busy writing a RESTful API in flask using flask views to drive the API. Since the application is sizable, I'm modularizing the code into blueprints. I've run into some trouble when trying to catch signals in a blueprint. I can happily write the 'connect signal' code in my main __init__.py file but I'd like to write the signal listeners in the relevant blueprint so as not to have the main create_app method cluttered with blueprint specific code.

Currently I have this [simplified] working code:

def create_app(debug=False):
    app = Flask(__name__)
    ...
    app.register_blueprint(my_blueprint)
    @mysignal.connect_via(app)
    def print_howdy(sender, **extra):
       print "howdy"

I'd like to move the signal connection code into my_blueprint, but cannot find a way to do that elegantly. How can I make this work?

Answer 1

Remember that the decorator syntax in Python is sugar:

@decorator
def decorated():
    pass

is equivalent to:

def decorated():
    pass

decorated = decorator(decorated)

Therefore, you can define print_howdy in your blueprint and import it along with your blueprint. In order to keep things clean, you can even have a list of signal listeners defined in your blueprint and import them all in one go:

mysignal.connect_via(app)(print_howdy)

or even:

# blueprint.py
def print_howdy(): pass
def print_seeya(): pass

MYSIGNAL_LISTENERS = (print_howdy, print_seeya)

# __init__.py
from blueprint import MYSIGNAL_LISTENERS
for listener in MYSIGNAL_LISTENERS:
    mysignal.connect_via(app)(listener)

You can even abstract this pattern into a function:

from werkzeug.utils import import_string

DEFAULT_SIGNALS = ('MYSIGNAL', 'MYOTHERSIGNAL')

def register_blueprint_and_signals(app, bp_path, signals=DEFAULT_SIGNALS):
    bp = import_string(bp_path)
    app.register_blueprint(bp)


    signal_path = bp_path.rsplit(".", 1)[0] + "."
    for signal in signals:
        try:
            listeners = import_string(signal_path + signal + '_LISTENERS')
        except ImportError:
            continue

        for listener in listeners:
            signal.connect_via(app)(listener)

Which you could then use like so:

app = Flask(__name__)
register_blueprint_and_signals(app, "my_module.my_blueprint")
register_blueprint_and_signals(app, "another.blueprint", ['ANOTHERSIGNAL'])