Shulou(Shulou.com)06/01 Report--

This article focuses on "how to use Python's blinker library". Interested friends may wish to have a look. The method introduced in this paper is simple, fast and practical. Let's let the editor take you to learn how to use Python's blinker library.

1. Signal

A signal is a way of notification or communication, and the signal is divided into sender and receiver. The sender sends a signal, and the process received by the receiver jumps into the signal processing function, and then jumps back to the original position to continue execution.

The common signal in Linux, entering Ctrl+C through the keyboard, is to send a signal to the system to tell the system to exit the current process.

The characteristic of the signal is that the sender informs the subscriber what has happened. The use of signals is divided into three steps: defining the signal, listening to the signal, and sending the signal.

The communication module of signal concept is provided in Python, which is blinker.

Blinker is a powerful signal library based on Python, which supports not only simple point-to-point communication, but also point-to-multipoint multicast. The signaling mechanism of Flask is based on it. Although the kernel of Blinker is small, it is very powerful. It supports the following features:

Support for registering global named signals

Anonymous signals are supported

Support for custom named signals

Support persistent and transient connections to the recipient

Automatic disconnection from the recipient through weak references

Support for sending data of any size

Support to collect the return value of the signal receiver

Thread safety

2. Use of blinker

Installation method:

Pip install blinker2.1 named signal from blinker import signal # defines a signal s = signal ('king') def animal (args): print (' I am Xiao Dengfeng, the king is back, I'm going to tour the mountains') # signal registers a receiver s.connect (animal) if "_ _ main__" = = _ _ name__: # send signal s.send ()

2.2 Anonymous signal

Blinker also supports anonymous signals, but there is no need to specify a specific signal value. Each anonymous signal created is independent of each other.

From blinker import Signal s = Signal () def animal (sender): print ('I'm Xiaofeng, the king is back, I'm going to patrol the mountains') s.connect (animal) if "_ _ main__" = _ _ name__: s.send () 2.3Multicast signal

Multicast signal is a characteristic that can reflect the advantages of the signal. Multiple receivers register with the signal, and the sender only needs to send the message once to transmit the information to multiple receivers.

From blinker import signal s = signal ('king') def animal_one (args): print (f' I am a small drill wind, today's slogan is: {args}') def animal_two (args): print (I am a big drill wind, today's slogan is: {args}') s.connect (animal_one) s.connect (animal_two) if "_ _ main__" = _ _ name__: s.send ('the king asked me to patrol the mountains Grab a monk for dinner!')

2.4 recipients subscribe to topics

The recipient supports subscribing to the specified topic and sends it to the recipient only when the specified topic sends a message. This approach makes a good distinction between different themes.

From blinker import signal s = signal ('king') def animal (args): print (f' I am drillwind, {args} is my eldest brother') s.connect (animal, sender=' Elephant') if "_ main__" = = _ name__: for i in ['Lion', 'Elephant', 'Dapeng']: s.send (I)

2.5 usage of decorator

In addition to function registration, there is a simpler method of signal registration, and that is the decorator.

From blinker import signal s = signal ('king') @ s.connectdef animal_one (args): print (I am a small drill wind, today's slogan is: {args}') def animal_two (args): print (I am a big drill wind, today's slogan is: {args}') if "_ main__" = _ _ name__: s.send ('the king asked me to patrol the mountains and grab a monk to make dinner!') 2.6 Decoration for subscriptions to topics

One drawback of connect's registration method using decorators is that you cannot subscribe to topics, so there are more advanced connect_via methods that support subscribing to topics.

From blinker import signals = signal ('king') @ s.connect_via (' Elephant') def animal (args): print (f'I am a drillwind, {args} is my eldest brother') if "_ _ main__" = = _ name__: for i in ['Lion', 'Elephant', 'Dapeng']: s.send (I) 2.7 check whether the signal has a receiver

If it takes a long time for a sender to prepare before sending a message, in order to avoid wasting performance without a receiver, you can first check whether a signal has a receiver and send it only when it is certain that there is a receiver. Be accurate.

From blinker import signals = signal ('king') Q = signal (' queue') def animal (sender): print ('I am Xiao Dengfeng, the king is back S.connect (animal) if "_ _ main__" = = _ _ name__: res = s.receivers print (res) if res: s.send () res = q.receivers q.send () else: print ("the kids are out on a mountain tour")

{4511880240:}

I'm Xiao Dengfeng. The king is back. I'm going to patrol the mountains.

{}

The children are out on a mountain tour.

2.8 check whether the subscriber subscribes to a certain signal

You can also check whether the subscriber is caused by a certain signal

From blinker import signals = signal ('king') Q = signal (' queue') def animal (sender): print ('I am Xiaofeng, the king is back, I'm going to tour the mountains') s.connect (animal) if "_ main__" = _ _ name__: res = s.has_receivers_for (animal) print (res) res = q.has_receivers_for (animal)

True

False

3. Flask signal based on blinker

Flask integrates blinker as a solution for decoupling applications. In Flask, signal usage scenarios such as before the request arrives and after the request ends. Flask also supports custom signals.

3.1simple Flask demofrom flask import Flaskapp = Flask (_ _ name__) @ app.route ('/', methods= ['GET','POST'], endpoint='index') def index (): return' hello blinker' if _ _ name__ = ='_ main__': app.run ()

When you access 127.0.0.1 hello blinker 5000, return it to the browser.

3.2 Custom signal

Because Flask integrates signals, it is introduced from Flask when signals are used in Flask.

From flask import Flaskfrom 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 ()

3.3 Flask built-in signal

In addition to customizing the signal in Flask, you can also use your own signal. There are many kinds of signals that come with Flask, as follows:

Request request_started = _ signals.signal ('request-started') # execute request_finished = _ signals.signal (' request-finished') # perform template rendering after the request ends before_render_template = _ signals.signal ('before-render-template') # execute template_rendered = _ signals.signal (' template-rendered') before template rendering # request for execution after template rendering got_request_exception = _ signals.signal ('got-request-exception') # execute when an exception occurs in the request execution request_tearing_down = _ signals.signal (' request-tearing-down') # automatically execute after the request is completed (whether successful or not) appcontext_tearing_down = _ signals.signal ('appcontext-tearing-down') # the context of the request is finished Automatic execution after completion (whether successful or not) appcontext_pushed = _ signals.signal ('appcontext-pushed') # execute appcontext_popped when requesting context push = _ signals.signal (' appcontext-popped') # execute message_flashed when requesting context pop = _ signals.signal ('message-flashed') # call flask to add data to it Automatic touch

Let's take a look at how the signal in Flask is used before the request arrives.

From flask import Flaskfrom flask.signals import _ signals, request_startedimport time app = Flask (_ _ name__) def wechat (args): print ('you have msg from wechat') # introduce the determined signal from flask and register a function request_started.connect (wechat) @ app.route (' /', methods= ['GET','POST'], endpoint='index') def index (): return' hello blinker'if _ name__ = ='_ main__': app.run ()

When the request arrives, Flask notifies the recipient via request_started, which is the function wechat, and the wechat function executes first and then returns the result to the browser.

However, this method is not very authentic, because the signal does not support asynchronous methods, so the receiver of the signal in the production environment is usually configured with asynchronous execution framework, such as the famous asynchronous framework celery in Python.

At this point, I believe you have a deeper understanding of "how to use Python's blinker library". You might as well do it in practice. Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.