Coverage for src/zapy/utils/singleton.py: 100%

1from __future__ import annotations 

2 

3from threading import Lock 

4from typing import Any, Generic, TypeVar 

5 

6_T = TypeVar("_T") 

7 

8 

9class SingletonMeta(type, Generic[_T]): 

10 """ 

11 This is a thread-safe implementation of Singleton. 

12 """ 

13 

14 _instances: dict[SingletonMeta[_T], _T] = {} # noqa: RUF012 

15 

16 _lock: Lock = Lock() 

17 """ 

18 We now have a lock object that will be used to synchronize threads during 

19 first access to the Singleton. 

20 """ 

21 

22 def __call__(cls, *args: Any, **kwargs: Any) -> _T: 

23 """ 

24 Possible changes to the value of the `__init__` argument do not affect 

25 the returned instance. 

26 """ 

27 # Now, imagine that the program has just been launched. Since there's no 

28 # Singleton instance yet, multiple threads can simultaneously pass the 

29 # previous conditional and reach this point almost at the same time. The 

30 # first of them will acquire lock and will proceed further, while the 

31 # rest will wait here. 

32 with cls._lock: 

33 # The first thread to acquire the lock, reaches this conditional, 

34 # goes inside and creates the Singleton instance. Once it leaves the 

35 # lock block, a thread that might have been waiting for the lock 

36 # release may then enter this section. But since the Singleton field 

37 # is already initialized, the thread won't create a new object. 

38 if cls not in cls._instances: 

39 instance = super().__call__(*args, **kwargs) 

40 cls._instances[cls] = instance 

41 return cls._instances[cls]