Using Blinker for Signal Handling in Python and Flask

Signals are a communication mechanism where a sender emits a notification that receivers handle via a callback function. In Linux, pressing Ctrl+C sends a termination signal to a process.

Python provides the blinker library for signal handling. Blinker supports global named signals, anonymous signals, custom naming, persistent and weak connections, arbitrary data payloads, receiver return collection, and thread safety.

2. Blinker Usage

2.1 Named Signal

<code>from blinker import signal

# Define a signal
s = signal('king')

def animal(args):
    print('我是小钻风，大王回来了，我要去巡山')

# Register a receiver
s.connect(animal)

if "__main__" == __name__:
    # Send the signal
    s.send()
</code>

2.2 Anonymous Signal

<code>from blinker import Signal

s = Signal()

def animal(sender):
    print('我是小钻风，大王回来了，我要去巡山')

s.connect(animal)

if "__main__" == __name__:
    s.send()
</code>

2.3 Multicast Signal

<code>from blinker import signal

s = signal('king')

def animal_one(args):
    print(f'我是小钻风，今天的口号是: {args}')

def animal_two(args):
    print(f'我是大钻风，今天的口号是: {args}')

s.connect(animal_one)
s.connect(animal_two)

if "__main__" == __name__:
    s.send('大王叫我来巡山，抓个和尚做晚餐！')
</code>

2.4 Topic Subscription

<code>from blinker import signal

s = signal('king')

def animal(args):
    print(f'我是小钻风，{args} 是我大哥')

s.connect(animal, sender='大象')

if "__main__" == __name__:
    for i in ['狮子', '大象', '大鹏']:
        s.send(i)
</code>

2.5 Decorator Registration

<code>from blinker import signal

s = signal('king')

@s.connect
def animal_one(args):
    print(f'我是小钻风，今天的口号是: {args}')

@s.connect
def animal_two(args):
    print(f'我是大钻风，今天的口号是: {args}')

if "__main__" == __name__:
    s.send('大王叫我来巡山，抓个和尚做晚餐！')
</code>

2.6 Decorator with Topic

<code>from blinker import signal

s = signal('king')

@s.connect_via('大象')
def animal(args):
    print(f'我是小钻风，{args} 是我大哥')

if "__main__" == __name__:
    for i in ['狮子', '大象', '大鹏']:
        s.send(i)
</code>

2.7 Checking for Receivers

<code>from blinker import signal

s = signal('king')
q = signal('queue')

def animal(sender):
    print('我是小钻风，大王回来了，我要去巡山')

s.connect(animal)

if "__main__" == __name__:
    res = s.receivers
    print(res)
    if res:
        s.send()
    res = q.receivers
    print(res)
    if res:
        q.send()
    else:
        print('孩儿们都出去巡山了')
</code>

2.8 Checking Specific Subscription

<code>from blinker import signal

s = signal('king')
q = signal('queue')

def animal(sender):
    print('我是小钻风，大王回来了，我要去巡山')

s.connect(animal)

if "__main__" == __name__:
    res = s.has_receivers_for(animal)
    print(res)
    res = q.has_receivers_for(animal)
    print(res)
</code>

3. Flask Integration with Blinker

3.1 Simple Flask Demo

<code>from flask import Flask

app = Flask(__name__)

@app.route('/', methods=['GET','POST'], endpoint='index')
def index():
    return 'hello blinker'

if __name__ == '__main__':
    app.run()
</code>

Visiting 127.0.0.1:5000 returns hello blinker .

3.2 Custom Flask Signal

<code>from flask import Flask
from flask.signals import _signals

app = Flask(__name__)

s = _signals.singal('msg')

def QQ(args):
    print('you have msg from QQ')

s.connect(QQ)

@app.route('/', methods=['GET','POST'], endpoint='index')
def index():
    s.send()
    return 'hello blinker'

if __name__ == '__main__':
    app.run()
</code>

3.3 Built‑in Flask Signals

Flask defines many signals such as request_started , request_finished , before_render_template , template_rendered , got_request_exception , request_tearing_down , appcontext_tearing_down , appcontext_pushed , appcontext_popped , and message_flashed .

Example using request_started :

<code>from flask import Flask
from flask.signals import _signals, request_started
import time

app = Flask(__name__)

def wechat(args):
    print('you have msg from wechat')

request_started.connect(wechat)

@app.route('/', methods=['GET','POST'], endpoint='index')
def index():
    return 'hello blinker'

if __name__ == '__main__':
    app.run()
</code>

When a request arrives, Flask emits request_started , invoking the wechat function before returning the response.

4. Summary

Advantages of signals include decoupling applications and enabling publish‑subscribe patterns that notify multiple receivers with a single emission.

Drawbacks are lack of asynchronous support and limited ability to subscribe to specific topics.