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

1. Is this correct for python, or is the pattern wrong if so what would be the norm in this situation

2. How do I handle typing for AbstractService if my solution is correct (for any function that requires that type to be passed to it)

CodePudding user response：

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:
        ...