Simple example of my problem:
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'})
The issue is that all services in this instance will share subscribers, hence my solution was to create a function that returns the abstract service as such:
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'})
This now works because subscribers are unique to each class inheriting the abstract service.
Issues
-
Is this correct for python, or is the pattern wrong if so what would be
the norm in this situation -
How do I handle typing for AbstractService if my solution is correct
(for any function that requires that type to be passed to it)
>Solution :
i would approach it like this:
from __future__ import annotations
class AbstractService:
subscribers: list[AbstractService]
def __init_subclass__(cls, **kw):
super().__init_subclass__(**kw)
cls.subscribers = []
it sets a new subscribers instance on each class after creation.
edit: additionally this handles typing for you as well by using from __future__ import annotations
e.g.
class C:
def f(self) -> C:
...