flask中使用信号来执行异步任务

比如在接口中执行某些操作之后，执行异步通知的操作但不使用消息队列或redis，可以通过信号的方式来做。

from flask.signals import Namespace
_signal = Namespace()

apply_task_signal = _signal.signal(name='some', doc='')

def apply_task_send(action_type, action_method, action_data, namespace='default', **kwargs):
  
  apply_task_signal.send(current_app._get_current_object(),namespace=namespace,action_method=action_method, action_data=action_data,action_type=action_type, **kwargs)

def apply_task_signal_callback(sender,namespace, action_type,action_method,action_data,**kwargs):
    taskmanager.delay
    taskmanager.apply_async

class TaskManager:
    def __init__(self,namespace,action_type,action_method,action_data,*args,**kwargs):
        self.namespace = namespace
        self.action_type = action_type
        ...
        self.handler = getattr(self, self.action_type)()

    def __call__(self):
        return getattr(self.handler, self.action_method)(self.action_data)

    def typea(self):
        cls_handler = typea()
        return cls_handler

class typea():
    def method1(action_data):
        # 这样就把任务需要的数据传入到特定的函数中
        # 当然一般还需要一个装饰器检验传入的action_data数据是否正常，如果异常则需要日志记录
        pass
    
@celery.task
def taskmanager(namespace,action_type,action_method,action_data,**kwargs):
    t_manager = TaskManager(...)
    t_manager()

current_app 那是获取当前的应用上下文
TaskManager是通过传入的action_type method data来触发不同的异步任务的 管理器