SharedObjectsListPlugin.method() - Code Metrics - Inspection of "Merge branch 'recipes' of https://github.com/usebl..." - useblocks/groundwork - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 6fd567...f65269 )

by Daniel

created 2016-09-09 09:23 UTC

SharedObjectsListPlugin.method() A

↳ Parent: SharedObjectsListApplication.__init__()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	5
rs	9.4285

from groundwork.patterns.gw_base_pattern import GwBasePattern


class GwSharedObjectsPattern(GwBasePattern):
    """
    Pattern, which provides access to shared object functionality.

    Use shared objects to provide access for other plugins to objects, which are created by this plugin.
    Or use it to get access to objects provided by other plugins.

    Common use cases are stores, which handle business logic and database abstraction. E.g. a user store.

    Provided function:

     * self.shared_objects.register()
     * self.shared_objects.unregister()
     * self.shared_objects.get()

    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if not hasattr(self.app, "shared_objects"):
            self.app.shared_objects = SharedObjectsListApplication(self.app)
        self.shared_objects = SharedObjectsListPlugin(self)

    def activate(self):
        pass

    def deactivate(self):
        pass


class SharedObjectsListPlugin:

    """
    Cares about plugin function for shared objects on plugin level.

    The class mainly directs most function calls to the ShareObjectApplication class, which is initiated on
    application level.
    """
    def __init__(self, plugin):
        """
        :param plugin: initiated plugin object
        """
        self.plugin = plugin
        self.app = plugin.app
        self.log = plugin.log
        # Let's register a receiver, which cares about the deactivation process of shared objects for this plugin.
        # We do it after the original plugin deactivation, so we can be sure that the registered function is the last
        # one which cares about shared objects for this plugin.
        self.plugin.signals.connect(receiver="%s_shared_objects_deactivation" % self.plugin.name,
                                    signal="plugin_deactivate_post",
                                    function=self.__deactivate_shared_objects,
                                    description="Deactivate documents for %s" % self.plugin.name,
                                    sender=self.plugin)

        self.log.debug("Shared objects for plugin %s initialised" % self.plugin.name)

    def __deactivate_shared_objects(self, plugin, *args, **kwargs):
        """
        Callback, which gets executed, if the signal "plugin_deactivate_post" was send by the plugin.
        """
        shared_objects = self.get()
        for shared_object in shared_objects.keys():
            self.unregister(shared_object)

    def register(self, name, description, obj):
        """
        Registers a new shared object.

        :param name: Unique name for shared object
        :type name: str
        :param description: Description of shared object
        :type description: str
        :param obj: The object, which shall be shared
        :type obj: any type
        """
        return self.app.shared_objects.register(name, description, obj, self.plugin)

    def unregister(self, shared_object):
        """
        Unregisters an already registered shared object.

        :param shared_object: name of the shared object
        :type shared_object: str
        """
        return self.app.shared_objects.unregister(shared_object)

    def get(self, name=None):
        """
        Returns requested shared objects, which were registered by the current plugin.

        If access to objects of other plugins are needed, use :func:`access` or perform get on application level::

            my_app.shared_objects.get(name="...")

        :param name: Name of a request shared object
        :type name: str or None
        """
        return self.app.shared_objects.get(name, self.plugin)

    def access(self, name):
        """
        Returns the object of the shared_object, if the given name has been registered.
        The search is done on application level, so registered shared objects form other plugins
        can be access.

        :param name: Name of the shared object
        :return: object, whatever it may be...
        """
        return self.app.shared_objects.access(name)


class SharedObjectsListApplication:
    """
    Cares about shared objects on application level.
    """
    def __init__(self, app):
        """
        :param app: groundwork application object, for which shared objects shall be activated on application level.
        """
        self.app = app
        self.log = app.log
        self._shared_objects = {}
        self.log.debug("Application shared objects initialised")

    def get(self, name=None, plugin=None):
        """
        Returns requested shared objects.

        :param name: Name of a request shared object
        :type name: str or None
        :param plugin: Plugin, which has registered the requested shared object
        :type plugin: GwBasePattern instance or None
        """
        if plugin is not None:
            if name is None:
                shared_objects_list = {}
                for key in self._shared_objects.keys():
                    if self._shared_objects[key].plugin == plugin:

                        shared_objects_list[key] = self._shared_objects[key]
                return shared_objects_list
            else:
                if name in self._shared_objects.keys():
                    if self._shared_objects[name].plugin == plugin:
                        return self._shared_objects[name]
                    else:
                        return None
                else:
                    return None
        else:
            if name is None:
                return self._shared_objects
            else:
                if name in self._shared_objects.keys():
                    return self._shared_objects[name]
                else:
                    return None

    def access(self, name):
        """
        Returns the object of the shared_object, if the given name has been registered.

        Unlike :func:`get()`, which returns the complete instance of a shared object, including
        name, description, plugin, access() returns the object only (without any meta data).

        :param name: Name of the shared object
        :return: object, whatever it may be...
        """
        return self.get(name, plugin=None).obj

    def register(self, name, description, obj, plugin):
        """
        Registers a new shared object.

        :param name: Unique name for shared object
        :type name: str
        :param description: Description of shared object
        :type description: str
        :param obj: The object, which shall be shared
        :type obj: any type
        :param plugin: Plugin, which registers the new shared object
        """
        if name in self._shared_objects.keys():
            raise SharedObjectExistException("Shared Object %s already registered by %s"
                                             % (name, self._shared_objects[name].plugin.name))

        new_shared_object = SharedObject(name, description, obj, plugin)
        self._shared_objects[name] = new_shared_object
        self.log.debug("Shared object registered: %s" % name)
        return new_shared_object

    def unregister(self, shared_object):
        """
        Unregisters an existing shared object, so that this shared object is no longer available.

        This function is mainly used during plugin deactivation.

        :param shared_object: Name of the shared_object
        """
        if shared_object not in self._shared_objects.keys():
            self.log.warning("Can not unregister shared object %s" % shared_object)
        else:
            del (self._shared_objects[shared_object])
            self.log.debug("Shared object %s got unregistered" % shared_object)


class SharedObject:
    """
    SharedObject class, which stores the objects itself besides meta data like an unique name, a description and the
    plugin, which registers the shared object.
    """
    def __init__(self, name, description, obj, plugin):
        """
        :param name: Unique name of the shared object
        :param description: Description
        :param obj: The object itself, which gets shared
        :param plugin: The plugin, which registered this shared object
        """
        self.name = name
        self.description = description
        self.obj = obj
        self.plugin = plugin


class SharedObjectExistException(BaseException):
    pass


1		from groundwork.patterns.gw_base_pattern import GwBasePattern
2
3
4		class GwSharedObjectsPattern(GwBasePattern):
5		"""
6		Pattern, which provides access to shared object functionality.
7
8		Use shared objects to provide access for other plugins to objects, which are created by this plugin.
9		Or use it to get access to objects provided by other plugins.
10
11		Common use cases are stores, which handle business logic and database abstraction. E.g. a user store.
12
13		Provided function:
14
15		* self.shared_objects.register()
16		* self.shared_objects.unregister()
17		* self.shared_objects.get()
18
19		"""
20
21		def __init__(self, args, *kwargs):
22		super().__init__(args, *kwargs)
23		if not hasattr(self.app, "shared_objects"):
24		self.app.shared_objects = SharedObjectsListApplication(self.app)
25		self.shared_objects = SharedObjectsListPlugin(self)
26
27		def activate(self):
28		pass
29
30		def deactivate(self):
31		pass
32
33
34	View Code Duplication	class SharedObjectsListPlugin:
		0 ignored issues – show Duplication introduced 2016-09-09 10:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
35		"""
36		Cares about plugin function for shared objects on plugin level.
37
38		The class mainly directs most function calls to the ShareObjectApplication class, which is initiated on
39		application level.
40		"""
41		def __init__(self, plugin):
42		"""
43		:param plugin: initiated plugin object
44		"""
45		self.plugin = plugin
46		self.app = plugin.app
47		self.log = plugin.log
48		# Let's register a receiver, which cares about the deactivation process of shared objects for this plugin.
49		# We do it after the original plugin deactivation, so we can be sure that the registered function is the last
50		# one which cares about shared objects for this plugin.
51		self.plugin.signals.connect(receiver="%s_shared_objects_deactivation" % self.plugin.name,
52		signal="plugin_deactivate_post",
53		function=self.__deactivate_shared_objects,
54		description="Deactivate documents for %s" % self.plugin.name,
55		sender=self.plugin)
56
57		self.log.debug("Shared objects for plugin %s initialised" % self.plugin.name)
58
59		def __deactivate_shared_objects(self, plugin, args, *kwargs):
60		"""
61		Callback, which gets executed, if the signal "plugin_deactivate_post" was send by the plugin.
62		"""
63		shared_objects = self.get()
64		for shared_object in shared_objects.keys():
65		self.unregister(shared_object)
66
67		def register(self, name, description, obj):
68		"""
69		Registers a new shared object.
70
71		:param name: Unique name for shared object
72		:type name: str
73		:param description: Description of shared object
74		:type description: str
75		:param obj: The object, which shall be shared
76		:type obj: any type
77		"""
78		return self.app.shared_objects.register(name, description, obj, self.plugin)
79
80		def unregister(self, shared_object):
81		"""
82		Unregisters an already registered shared object.
83
84		:param shared_object: name of the shared object
85		:type shared_object: str
86		"""
87		return self.app.shared_objects.unregister(shared_object)
88
89		def get(self, name=None):
90		"""
91		Returns requested shared objects, which were registered by the current plugin.
92
93		If access to objects of other plugins are needed, use :func:`access` or perform get on application level::
94
95		my_app.shared_objects.get(name="...")
96
97		:param name: Name of a request shared object
98		:type name: str or None
99		"""
100		return self.app.shared_objects.get(name, self.plugin)
101
102		def access(self, name):
103		"""
104		Returns the object of the shared_object, if the given name has been registered.
105		The search is done on application level, so registered shared objects form other plugins
106		can be access.
107
108		:param name: Name of the shared object
109		:return: object, whatever it may be...
110		"""
111		return self.app.shared_objects.access(name)
112
113
114		class SharedObjectsListApplication:
115		"""
116		Cares about shared objects on application level.
117		"""
118		def __init__(self, app):
119		"""
120		:param app: groundwork application object, for which shared objects shall be activated on application level.
121		"""
122		self.app = app
123		self.log = app.log
124		self._shared_objects = {}
125		self.log.debug("Application shared objects initialised")
126
127		def get(self, name=None, plugin=None):
128		"""
129		Returns requested shared objects.
130
131		:param name: Name of a request shared object
132		:type name: str or None
133		:param plugin: Plugin, which has registered the requested shared object
134		:type plugin: GwBasePattern instance or None
135		"""
136		if plugin is not None:
137		if name is None:
138		shared_objects_list = {}
139		for key in self._shared_objects.keys():
140	View Code Duplication	if self._shared_objects[key].plugin == plugin:
		0 ignored issues – show Duplication introduced 2016-08-10 13:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
141		shared_objects_list[key] = self._shared_objects[key]
142		return shared_objects_list
143		else:
144		if name in self._shared_objects.keys():
145		if self._shared_objects[name].plugin == plugin:
146		return self._shared_objects[name]
147		else:
148		return None
149		else:
150		return None
151		else:
152		if name is None:
153		return self._shared_objects
154		else:
155		if name in self._shared_objects.keys():
156		return self._shared_objects[name]
157		else:
158		return None
159
160		def access(self, name):
161		"""
162		Returns the object of the shared_object, if the given name has been registered.
163
164		Unlike :func:`get()`, which returns the complete instance of a shared object, including
165		name, description, plugin, access() returns the object only (without any meta data).
166
167		:param name: Name of the shared object
168		:return: object, whatever it may be...
169		"""
170		return self.get(name, plugin=None).obj
171
172		def register(self, name, description, obj, plugin):
173		"""
174		Registers a new shared object.
175
176		:param name: Unique name for shared object
177		:type name: str
178		:param description: Description of shared object
179		:type description: str
180		:param obj: The object, which shall be shared
181		:type obj: any type
182		:param plugin: Plugin, which registers the new shared object
183		"""
184		if name in self._shared_objects.keys():
185		raise SharedObjectExistException("Shared Object %s already registered by %s"
186		% (name, self._shared_objects[name].plugin.name))
187
188		new_shared_object = SharedObject(name, description, obj, plugin)
189		self._shared_objects[name] = new_shared_object
190		self.log.debug("Shared object registered: %s" % name)
191		return new_shared_object
192
193		def unregister(self, shared_object):
194		"""
195		Unregisters an existing shared object, so that this shared object is no longer available.
196
197		This function is mainly used during plugin deactivation.
198
199		:param shared_object: Name of the shared_object
200		"""
201		if shared_object not in self._shared_objects.keys():
202		self.log.warning("Can not unregister shared object %s" % shared_object)
203		else:
204		del (self._shared_objects[shared_object])
205		self.log.debug("Shared object %s got unregistered" % shared_object)
206
207
208		class SharedObject:
209		"""
210		SharedObject class, which stores the objects itself besides meta data like an unique name, a description and the
211		plugin, which registers the shared object.
212		"""
213		def __init__(self, name, description, obj, plugin):
214		"""
215		:param name: Unique name of the shared object
216		:param description: Description
217		:param obj: The object itself, which gets shared
218		:param plugin: The plugin, which registered this shared object
219		"""
220		self.name = name
221		self.description = description
222		self.obj = obj
223		self.plugin = plugin
224
225
226		class SharedObjectExistException(BaseException):
227		pass
228

useblocks / groundwork

Push — master ( 6fd567...f65269 )

SharedObjectsListPlugin.method() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like