如何使用装饰器在 class 中正确注册处理程序？

Question

我正在实现一个消息处理器 class。我想使用装饰器来确定应该使用哪个 class 方法来处理特定的命令类型。 Flask 使用他们的 @app.route('/urlpath') 装饰器做了一些与此非常相似的事情。

我想不出一种方法将函数实际注册为处理程序，而不首先显式执行该函数，这似乎没有必要。

我的代码：

from enum import Enum, auto, unique
import typing

@unique
class CommandType(Enum):
    DONE = auto()
    PASS = auto()

class MessageProcessor:
    def __init__(self):
        self.handlers = {}

    def register_handler(command_type: CommandType, *args) -> typing.Callable:
        def decorator(function: typing.Callable):
            def wrapper(self, args):
                self.handlers[command_type] = function
                return function
            return wrapper
        return decorator

    @register_handler(CommandType.DONE)
    def handle_done(self, args):
        print("Handler for DONE command. args: ", args)

    @register_handler(CommandType.PASS)
    def handle_pass(self, args):
        print("Handler for PASS command. args: ", args)

    def process(self, message: str):
        tokens = message.split()
        command_type = CommandType[tokens[0]]
        args = tokens[1:]

        self.handlers[command_type](self, args)

现在，如果我尝试执行以下代码，我会收到错误消息：

m = MessageProcessor()
m.process('DONE arg1 arg2')

Traceback (most recent call last):
  File "/Users/......./main.py", line 87, in <module>
    m.process('DONE arg1 arg2')
  File "/Users/......./main.py", line 81, in process
    self.handlers[command_type](self, args)
KeyError: <CommandType.DONE: 1>

但是，如果我首先显式调用这些方法（带有任何参数），那么我不会收到错误消息：

m = MessageProcessor()
m.handle_done(None)
m.handle_pass(None)
m.process('DONE arg1 arg2')
m.process('PASS arg1 arg2')

Handler for DONE command. args:  ['arg1', 'arg2']
Handler for PASS command. args:  ['arg1', 'arg2']

有没有一种方法可以正确注册这些处理程序而无需首先执行显式虚拟调用（即 m.handle_done(None)）？

我不想在 MessageProcessor 的构造函数中手动调用每个方法，我正在寻找一种装饰器的存在就足够的方法。

我可以通过使处理程序静态化并且不绑定到对象实例来解决这个问题吗？我早些时候尝试过，但在尝试让它工作时遇到了麻烦：

class MessageProcessor:
    cls_handlers = {}

    def register_handler(command_type: CommandType, *args) -> typing.Callable:
        def decorator(function: typing.Callable):
            MessageProcessor.cls_handlers[command_type] = function
            return function
        return decorator

Answer 1

self.handlers[command_type] = function 问题：

register_handler decorator wrapper在函数执行时调用，在实例上注册handler，所以每个实例都需要调用每个handler。

MessageProcessor.cls_handlers[command_type] = function 问题：

register_handler decorator 在定义 class 时被调用，所以 MessageProcessor 还没有定义。

快速修复

在register_handler函数定义中引用cls_handlers：

# def register_handler(command_type: CommandType) -> typing.Callable:                               # -
def register_handler(command_type: CommandType, handlers: dict = cls_handlers) -> typing.Callable:  # +
    def decorator(function: typing.Callable):
        # MessageProcessor.cls_handlers[command_type] = function  # -
        handlers[command_type] = function                         # +
        return function
    return decorator

正确的方法

使用 class 之外的工厂创建 register_handler 函数：

def make_register_handler(handlers: dict):
    def register_handler(command_type: CommandType) -> typing.Callable:
        def decorator(function: typing.Callable):
            handlers[command_type] = function
            return function
        return decorator
    return register_handler

用法：

class MessageProcessor:
    cls_handlers = {}
    register_handler = make_register_handler(cls_handlers)

Flask中使用的方式

在 Flask 中，@app.route() 在 app 的 class 定义之外调用。

同样，处理程序可以在处理器之外定义 class:

class MessageProcessor:
    cls_handlers = {}

    @classmethod
    def register_handler(cls, command_type: CommandType) -> typing.Callable:
        def decorator(function: typing.Callable):
            cls.cls_handlers[command_type] = function
            return function
        return decorator

    def process(self, message: str):
        tokens = message.split()
        command_type = CommandType[tokens[0]]
        args = tokens[1:]

        self.cls_handlers[command_type](args)


@MessageProcessor.register_handler(CommandType.DONE)
def handle_done(args):
    print("Handler for DONE command. args: ", args)


@MessageProcessor.register_handler(CommandType.PASS)
def handle_pass(args):
    print("Handler for PASS command. args: ", args)

如何使用装饰器在 class 中正确注册处理程序？

How to properly register a handler within a class using a decorator?

python

flask

handler

python-decorators

快速修复

正确的方法

Flask中使用的方式