Mastering Python Signals with Blinker and Flask: A Complete Guide
This article introduces the concept of signals, explains the features of the Python Blinker library, demonstrates various usage patterns—including named, anonymous, multicast, and topic‑based signals—as well as how to integrate Blinker with Flask for custom and built‑in Flask signals, highlighting advantages and limitations.
1 Signal
Signal is a notification mechanism consisting of a sender and a receiver. When the sender emits a signal, the receiver's handler is invoked, runs, then returns control.
In Linux, pressing Ctrl+C sends a termination signal to the process.
Signals follow three steps: define, listen, and send.
Python provides a signal communication module called blinker.
Blinker is a powerful Python signal library supporting point‑to‑point and point‑to‑many communication. Flask’s signal system is built on it. Its core is small but offers many features:
Global named signals
Anonymous signals
Custom named signals
Persistent and transient connections
Automatic disconnection via weak references
Sending data of arbitrary size
Collecting receiver return values
Thread safety
2 Blinker Usage
Installation:
pip install blinker2.1 Named Signal
from blinker import signal
# define a signal
s = signal('king')
def animal(args):
print('I am a little drill, the king has returned, I will patrol the mountain')
# register receiver
s.connect(animal)
if __name__ == "__main__":
# send signal
s.send()2.2 Anonymous Signal
Blinker also supports anonymous signals, each independent.
from blinker import Signal
s = Signal()
def animal(sender):
print('I am a little drill, the king has returned, I will patrol the mountain')
s.connect(animal)
if __name__ == "__main__":
s.send()2.3 Multicast Signal
Multicast signals allow multiple receivers to be notified with a single send.
from blinker import signal
s = signal('king')
def animal_one(args):
print(f'I am little drill, today’s slogan: {args}')
def animal_two(args):
print(f'I am big drill, today’s slogan: {args}')
s.connect(animal_one)
s.connect(animal_two)
if __name__ == "__main__":
s.send('The king calls me to patrol, catch a monk for dinner!')2.4 Receiver Subscribes to Topic
Receivers can subscribe to a specific topic; they only receive messages sent with that topic.
from blinker import signal
s = signal('king')
def animal(args):
print(f'I am little drill, {args} is my brother')
s.connect(animal, sender='elephant')
if __name__ == "__main__":
for i in ['lion', 'elephant', 'roc']:
s.send(i)2.5 Decorator Usage
Functions can be registered as receivers using a decorator.
from blinker import signal
s = signal('king')
@s.connect
def animal_one(args):
print(f'I am little drill, today’s slogan: {args}')
@s.connect
def animal_two(args):
print(f'I am big drill, today’s slogan: {args}')
if __name__ == "__main__":
s.send('The king calls me to patrol, catch a monk for dinner!')2.6 Topic‑aware Decorator
The connect_via decorator allows subscribing to a specific topic.
from blinker import signal
s = signal('king')
@s.connect_via('elephant')
def animal(args):
print(f'I am little drill, {args} is my brother')
if __name__ == "__main__":
for i in ['lion', 'elephant', 'roc']:
s.send(i)2.7 Check for Receivers Before Sending
To avoid unnecessary work, you can check whether a signal has any receivers before sending.
from blinker import signal
s = signal('king')
q = signal('queue')
def animal(sender):
print('I am a little drill, the king has returned, I will patrol the mountain')
s.connect(animal)
if __name__ == "__main__":
if s.receivers:
s.send()
if q.receivers:
q.send()
else:
print('All children are out patrolling')2.8 Check If a Specific Receiver Is Subscribed
You can also verify whether a particular receiver is subscribed to a signal.
from blinker import signal
s = signal('king')
q = signal('queue')
def animal(sender):
print('I am a little drill, the king has returned, I will patrol the mountain')
s.connect(animal)
if __name__ == "__main__":
print(s.has_receivers_for(animal))
print(q.has_receivers_for(animal))3 Flask Signals Based on Blinker
Flask integrates Blinker to decouple applications. Common Flask signal use cases include before request, after request, template rendering, etc., and Flask also supports custom signals.
3.1 Simple Flask Demo
from flask import Flask
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'], endpoint='index')
def index():
return 'hello blinker'
if __name__ == '__main__':
app.run()3.2 Custom Signal in Flask
Because Flask bundles Blinker, custom signals are created via Flask’s signal module.
from flask import Flask
from flask.signals import _signals
app = Flask(__name__)
s = _signals.signal('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()3.3 Built‑in Flask Signals
Flask provides many built‑in 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, message_flashed, etc.
Example: using the request_started signal.
from flask import Flask
from flask.signals import _signals, request_started
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()When a request arrives, Flask emits request_started, invoking the registered wechat function before returning the response. Note that signals are synchronous; in production they are often combined with asynchronous task queues like Celery.
4 Summary
Advantages of signals:
Decouples applications, turning tightly coupled serial execution into multi‑stage execution.
Publish‑subscribe model reduces caller complexity, notifying multiple subscribers with a single call.
Disadvantages of signals:
Do not support asynchronous execution.
Limited ability to subscribe to topics.
Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
MaGe Linux Operations
Founded in 2009, MaGe Education is a top Chinese high‑end IT training brand. Its graduates earn 12K+ RMB salaries, and the school has trained tens of thousands of students. It offers high‑pay courses in Linux cloud operations, Python full‑stack, automation, data analysis, AI, and Go high‑concurrency architecture. Thanks to quality courses and a solid reputation, it has talent partnerships with numerous internet firms.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.
