dimanche 9 juillet 2023

How to create singleton classes dynamically in python

I want to create classes dynamically such that : class is created as per the "name" provided by the user. This class should be singleton.

I want to implement some sort of messaging functionality where the communication channel (topic) with name "name" should be created only once (singleton), and several observers can be attached to that topic.

I tried to do something like below but not sure how to create singleton classes dynamically.

class GTopic:
    def register(self, observer):
        pass


class TopicA(GTopic):
    instance = None

    class __TopicA:
        def __init__(self):
            self._observers = []
            self.messages = []

        def register(self, observer):
            if observer not in self._observers:
                self._observers.append(observer)

        def send_message(self, msg):
            for observer in self._observers:
                observer.update(__class__.__name__, id(self), msg)
            self.messages.remove(msg)

        def send(self, msg):
            if not msg:
                return
            self.messages.append(msg)
            self.send_message(msg)

    def __new__(cls):
        if not TopicA.instance:
            TopicA.instance = TopicA.__TopicA()
            print("TopicA not present, creating new")
        print("TopicA already present")
        return TopicA.instance


class TopicB(GTopic):
    instance = None

    class __TopicB:
        def __init__(self):
            self._observers = []
            self.messages = []

        def register(self, observer):
            if observer not in self._observers:
                self._observers.append(observer)

        def send_message(self, msg):
            for observer in self._observers:
                observer.update(__class__.__name__, id(self), msg)
            self.messages.remove(msg)

        def send(self, msg):
            if not msg:
                return
            self.messages.append(msg)
            self.send_message(msg)

    def __new__(cls):
        if not TopicB.instance:
            TopicB.instance = TopicB.__TopicB()
            print("TopicB not present, creating new")
        print("TopicB already present")
        return TopicB.instance

"""
def __new__(cls):
    if not cls.instance:
        cls.instance = cls.cls()
        print(f"{cls} not present, creating new")
    print(f"{cls} already present")
    return cls.instance
"""

class GQueue:

    @staticmethod
    def get_topic(topic):
        if topic == "A":
            return TopicA()
        elif topic == "B":
            return TopicB()
        """
        else:
            Topic = type("Topic{}".format(topic), (TopicA,), {"instance": None, "__new__": __new__})
            return Topic()
        """


class Observer:
    def __init__(self, a=0):
        self.a = a

    def update(self, topic_name, topic_id, msg):
        print(f"Message for observer {id(self)} from {topic_name}: {msg}")
        self.a = 1


if __name__ == "__main__":
    # client code
    topic_a = GQueue.get_topic("A")
    topic_b = GQueue.get_topic("B")
    obs1 = Observer()
    obs2 = Observer()
    topic_a.register(obs1)
    topic_b.register(obs1)
    topic_a.register(obs2)
    topic_a.send("hello to A")
    topic_b.send("hello to B")

    # Now I want to do something like :

    """
    topic_x = GQueue.get_topic("X")
    topic_x.register(obs1)
    """

Above code works fine. Now I want to do something like :

    topic_x = GQueue.get_topic("X")
    topic_x.register(obs1)

Aucun commentaire:

Enregistrer un commentaire