我的问题的简单例子:
class AbstractService:
subscribers = []
@classmethod
def subscribe(cls, service):
service.subscribers.append(cls)
@classmethod
@abstractmethod
def notify(cls, payload):
raise NotImplementedError(f"{cls.__name__} not implemented notify()")
class ServiceA(AbstractService):
@classmethod
def notify(cls, payload):
pass
class ServiceB(AbstractService):
@classmethod
def notify(cls, payload):
pass
ServiceB.subscribe(ServiceA)
ServiceA.notify({'message': 'hello'})
问题是,这个实例中的所有服务都将共享订阅服务器,因此,我的解决方案是创建一个函数,以返回抽象服务:
def get_abstract_service():
class AbstractService:
subscribers = []
@classmethod
def subscribe(cls, service):
service.subscribers.append(cls)
@classmethod
@abstractmethod
def notify(cls, payload):
raise NotImplementedError(f"{cls.__name__} not implemented notify()")
return AbstractService
class ServiceA(get_abstract_service()):
@classmethod
def notify(cls, payload):
pass
class ServiceB(get_abstract_service()):
@classmethod
def notify(cls, payload):
pass
ServiceB.subscribe(ServiceA)
ServiceA.notify({'message': 'hello'})
这现在起作用了,因为订阅服务器对于继承抽象服务的每个类都是唯一的。
发行
对于python,
的规范是什么?
如果我的解决方案是正确的(对于任何需要将该类型传递给
发布于 2022-03-17 08:50:38
我会这样对待它:
from __future__ import annotations
class AbstractService:
subscribers: list[AbstractService]
def __init_subclass__(cls, **kw):
super().__init_subclass__(**kw)
cls.subscribers = []
它在创建后在每个类上设置一个新的subscribers
实例。
编辑:此外,这还可以通过使用from __future__ import annotations
为您处理输入。
例如:
class C:
def f(self) -> C:
...
https://stackoverflow.com/questions/71515835
复制相似问题