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)
Advertisement
Answer
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: ...