信号（Signal）

前言

信号用于在特定事件发生时触发自定义函数。Flask 使用第三方库 Blinker 实现，Flask 2.3.2+ 会自动安装：

python -m pip install blinker

自定义信号

from datetime import datetime

from blinker import Namespace

ns = Namespace()
user_login = ns.signal("user_login")


@user_login.connect
def log_user_login(user):
    log_entry = f"User {user.username} logged in at {datetime.now()}"
    print(log_entry)


class User:
    def __init__(self, username):
        self.username = username


user = User("zhangsan")
user_login.send(user)

在 Flask 视图中使用信号

from blinker import Namespace
from flask import Flask, request

ns = Namespace()
sig_command = ns.signal("command")

app = Flask(__name__)


@sig_command.connect
def handle_command_a(command: str):
    print(f"ReceivedA command: {command}")


@sig_command.connect
def handle_command_b(command: str):
    print(f"ReceivedB command: {command}")


@app.post("/command")
def process_command():
    req_body = request.get_json()
    sig_command.send(req_body.get("command"))
    return "OK"

提示

同一个信号可以订阅多个函数，发送时全部触发。