Coverage for src/zapy/utils/singleton.py: 100%
14 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
1from __future__ import annotations
3from threading import Lock
4from typing import Any, Generic, TypeVar
6_T = TypeVar("_T")
9class SingletonMeta(type, Generic[_T]):
10 """
11 This is a thread-safe implementation of Singleton.
12 """
14 _instances: dict[SingletonMeta[_T], _T] = {} # noqa: RUF012
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 """
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]