Create a queue for each singleton Object

Question

I am trying to create singleton instance for each queue. If the instance is already created it should get connected to same instance or else create a new one. I have used the queue to create the queues.

from queue import Queue

class Singleton(Queue):
    __instance = None

    def __new__(cls):
        if cls.__instance is None:
            cls.__instance = super().__new__(cls)        
        return cls.__instance
    


class Queue(object):

    def __init__(self,**kwargs):
        self.queue = Singleton
        print(id(self.queue))

class Interpretor(object):

    def __init__(self,**kwargs):
        self.queue = Singleton
        print(id(self.queue))

I am getting the same queue id for both the queues

2611032512096
2611032512096

I think, there is a misunderstanding. Singleton means unique instance and you want to build 2 instances ? So first question, why Singleton ? Then, your Singleton seems to be good but in the same process your class attribut `__instance` will be shared. If you create another Process (using multiprocessing library) another PID will be created and `__instance` class attribut will not be shared in this case, so this will result by two instances created (one in each process). — Adrien Derouene, Dec 05 '22 at 13:39
I want to create singleton instances for each queue.I want to call the singleton instance with respect to the queue name/id. Since the instance is already created, if queue 1 is called it should go to the instance 1 and if queue 2 is called it should go to the 2nd instance. — lolxtimes, Dec 06 '22 at 05:28

score 0 · Answer 1 · answered Dec 06 '22 at 16:27

To solve your problem, you have to implement two Singleton with their own class attribut. An example is shown below (with the references of threads where I found help to this solution):

from queue import Queue

class C_Queue(Queue):
    __instance = None

    def __new__(cls):
        if cls.__instance is None:
            cls.__instance = super().__new__(cls)        
            cls.__instance.__initialized = False
        return cls.__instance

    def __init__(self,**kwargs):
        if self.__initialized:
            return
        self.initialized = True  # To not call twice super().__init__()
        super().__init__()
        print('queue1')

class C_Interpretor(Queue):
    __instance = None

    def __new__(cls):
        if cls.__instance is None:
            cls.__instance = super().__new__(cls)        
            cls.__instance.__initialized = False
        return cls.__instance

    def __init__(self,**kwargs):
        if self.__initialized:
            return
        self.initialized = True  # To not call twice super().__init__()
        super().__init__()
        print('queue2')

if __name__ == '__main__':
    q1 = C_Queue()
    q1.put(item='hey_1')
    q2 = C_Queue()
    q2.put(item='hey_2')
    i1 = C_Interpretor()
    i2 = C_Interpretor()
    print(id(q1))
    print(id(q2))
    print(q1.qsize())  # expected 2
    print(id(i1))
    print(id(i2))

More info can be found here:

Why is singleton class' __init__() method called twice?

issue with singleton python call two times __init__

Create a queue for each singleton Object

1 Answers1