1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
# From https://github.com/reyoung/singleton
from threading import Lock
class Singleton(object):
"""
The Singleton class decorator.
Like:
from singleton.singleton import Singleton
@Singleton
class IntSingleton(object):
def __init__(self):
pass
Use IntSingleton.instance() get the instance
"""
def __init__(self, cls):
"""
:param cls: decorator class type
"""
self.__cls = cls
self.__instance = None
def initialize(self, *args, **kwargs):
"""
Initialize singleton object if it has not been initialized
:param args: class init parameters
:param kwargs: class init parameters
"""
if not self.is_initialized():
self.__instance = self.__cls(*args, **kwargs)
def is_initialized(self):
"""
:return: true if instance is initialized
"""
return self.__instance is not None
def instance(self):
"""
Get singleton instance
:return: instance object
"""
if not self.is_initialized():
self.initialize()
return self.__instance
def __call__(self, *args, **kwargs):
"""
Disable new instance of original class
:raise TypeError:
"""
raise TypeError("Singletons must be access by instance")
def __instancecheck__(self, inst):
"""
Helper for isinstance check
"""
return isinstance(inst, self.__cls)
class ThreadSafeSingleton(object):
def __init__(self, cls):
self.__cls = cls
self.__instance = None
self.__mutex = Lock()
def is_initialized(self):
self.__mutex.acquire()
try:
return self.__instance is not None
finally:
self.__mutex.release()
def initialize(self, *args, **kwargs):
self.__mutex.acquire()
try:
if self.__instance is None:
self.__instance = self.__cls(*args, **kwargs)
finally:
self.__mutex.release()
def instance(self):
self.__mutex.acquire()
try:
if self.__instance is None:
self.__instance = self.__cls()
return self.__instance
finally:
self.__mutex.release()
def __call__(self, *args, **kwargs):
"""
Disable new instance of original class
:raise TypeError:
"""
raise TypeError("Singletons must be access by instance")
def __instancecheck__(self, inst):
"""
Helper for isinstance check
"""
return isinstance(inst, self.__cls)
|
