utils.py 文件源码

def responder(cls, plugin_name: str, *actions: [Enum],
                  queue: Enum=TaskQueue.SHORT):
        """
        Registers the decorated function as a responder to the actions
        provided. Specifying description as defaults on option specific args
        is mandatory.
        """
        def _wrapper(function):
            task = celery.task(function,
                               base=ExceptionLoggerTask,
                               queue=queue.value)
            for action in actions:
                cls._responders[action].append(task)
            cls._plugins[task] = plugin_name
            params = signature(function).parameters.values()
            cls._options[task] = [param.name for param in params
                                  if param.default is not Parameter.empty]
            return function
        return _wrapper