时间戳

首页 > 解决方案 > 在函数中定义一个类来中断装饰器的执行

python - 在函数中定义一个类来中断装饰器的执行

问题描述

我正在尝试在运行时配置装饰器。这与我之前的问题有些相关：How to configure a decorator in Python

这样做的动机是我试图“按原样”使用 Thespian 剧团代码。

在这里有这段代码是否合法，我在类方法中定义了类（因此称为装饰器）？同样，这样做的原因是我可以在max_count调用装饰器之前提供参数。

该模块是calculator.calculator（是的，也许是错误的选择）

class Scheduler:
    def __init__(self):
        self.actor_system = None

    def start(self):
        self.actor_system = ActorSystem('multiprocTCPBase')

    def stop(self):
        self.actor_system.shutdown()

    def launch(self, count, func_and_data, status_cb):
        class CalcPayload:
            def __init__(self, func_and_data, status_cb):
                self.func_and_data = func_and_data
                self.status_cb = status_cb

        @troupe(max_count=count)
        class Execute(ActorTypeDispatcher):
            def receiveMsg_CalcPayload(self, msg, sender):
                func = msg.func_and_data['func']
                data = msg.func_and_data['data']
                status_cb = msg.status_cb

                self.send(sender, func(data, status_cb))

        exec_actor = self.actor_system.createActor(Execute)

        for index in range(len(func_and_data)):
            calc_config = CalcPayload(func_and_data[index], status_cb)
            self.actor_system.tell(exec_actor, calc_config)

        for index in range(len(func_and_data)):
            result = self.actor_system.listen(timeout)

        self.actor_system.tell(exec_actor, ActorExitRequest())

由于各种原因，我在使用装饰器时无法将其应用到类中。在我引用的问题中有一个简短的讨论。

标签： pythonpython-decorators

解决方案

虽然不是无效的，但通常不建议将类定义为函数内部的局部变量，因为这会使函数外部难以访问该类。

相反，您可以在函数外部定义类，并在实际需要时通过使用类对象调用装饰器函数将装饰器函数应用于类：

class CalcPayload:
    def __init__(self, func_and_data, status_cb):
        self.func_and_data = func_and_data
        self.status_cb = status_cb


class Execute(ActorTypeDispatcher):
    def receiveMsg_CalcPayload(self, msg, sender):
        func = msg.func_and_data['func']
        data = msg.func_and_data['data']
        status_cb = msg.status_cb

        self.send(sender, func(data, status_cb))

class Scheduler:
    def __init__(self):
        self.actor_system = None

    def start(self):
        self.actor_system = ActorSystem('multiprocTCPBase')

    def stop(self):
        self.actor_system.shutdown()

    def launch(self, count, func_and_data, status_cb):
        exec_actor = self.actor_system.createActor(troupe(max_count=count)(Execute))

        for index in range(len(func_and_data)):
            calc_config = CalcPayload(func_and_data[index], status_cb)
            self.actor_system.tell(exec_actor, calc_config)

        for index in range(len(func_and_data)):
            result = self.actor_system.listen(timeout)

        self.actor_system.tell(exec_actor, ActorExitRequest())

推荐阅读