cdplib/cdplib/Singleton_Threadsafe.py

import functools
import threading

lock = threading.Lock()

def synchronized(lock):
    """ Synchronization decorator """
    def wrapper(f):
        @functools.wraps(f)
        def inner_wrapper(*args, **kw):
            with lock:
                return f(*args, **kw)
        return inner_wrapper
    return wrapper

class SingletonThreadsafe(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._locked_call(*args, **kwargs)
        return cls._instances[cls]

    @synchronized(lock)
    def _locked_call(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(SingletonThreadsafe, cls).__call__(*args, **kwargs)


class testClass(metaclass=SingletonThreadsafe):
    pass
    def __init__(self, message: str = None):
        self._message = message 

if __name__ == "__main__":
    x = testClass('hello')
    y = testClass('yellow')

    print(x==y)
    print(x._message)
    print(y._message)