isomer.events.system.generate_asyncapi() - Code Metrics - Inspection of "Merge remote-tracking branch 'origin/master'" - isomeric/isomer - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ef441d...0acd10 )

by Heiko 'riot'

created 2020-12-18 04:07 UTC

isomer.events.system.generate_asyncapi() B

↳ Parent: isomer.events.system

Complexity

Conditions

Size

Total Lines	37
Code Lines	25

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	25
nop	0
dl	0
loc	37
rs	8.3466
c	0
b	0
f	0

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

# Isomer - The distributed application framework
# ==============================================
# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""


Module: Events
==============

Major Isomer event declarations


"""

from copy import copy
from typing import Dict

from circuits.core import Event
from isomer.logger import isolog, events

# from isomer.ui.clientobjects import User


AuthorizedEvents: Dict[str, Event] = {}
AnonymousEvents: Dict[str, Event] = {}

populated = False


def get_user_events():
    """Return all registered authorized events"""

    return AuthorizedEvents


def get_anonymous_events():
    """Return all registered anonymous events"""

    return AnonymousEvents


def populate_user_events():
    """Generate a list of all registered authorized and anonymous events"""

    global AuthorizedEvents
    global AnonymousEvents
    global populated

    def inheritors(klass):
        """Find inheritors of a specified object class"""

        subclasses = {}
        subclasses_set = set()
        work = [klass]
        while work:
            parent = work.pop()
            for child in parent.__subclasses__():
                if child not in subclasses_set:
                    # pprint(child.__dict__)
                    name = child.__module__ + "." + child.__name__

                    subclasses_set.add(child)
                    event = {
                        "event": child,
                        "name": name,
                        "doc": child.__doc__,
                        "summary": child.summary,
                        "tags": child.tags,
                        "args": child.args,
                    }

                    if child.__module__ in subclasses:
                        subclasses[child.__module__][child.__name__] = event
                    else:
                        subclasses[child.__module__] = {child.__name__: event}
                    work.append(child)
        return subclasses

    # TODO: Change event system again, to catch authorized (i.e. "user") as
    # well as normal events, so they can be processed by Automat

    # NormalEvents = inheritors(Event)
    AuthorizedEvents = inheritors(authorized_event)
    AnonymousEvents = inheritors(anonymous_event)

    # AuthorizedEvents.update(NormalEvents)
    populated = True


class isomer_basic_event(Event):
    """Basic Isomer event class"""

    args = {}
    tags = []
    summary = "Basic Isomer Event"

    def __init__(self, *args, **kwargs):
        """Initializes a basic Isomer event.

        For further details, check out the circuits documentation.
        """
        super(isomer_basic_event, self).__init__(*args, **kwargs)


class isomer_ui_event(isomer_basic_event):
    """Isomer user interface event class"""

    pass


class isomer_event(isomer_basic_event):
    """Isomer internal event class"""

    pass


class anonymous_event(isomer_ui_event):
    """Base class for events for logged in users."""

    def __init__(self, action, data, client, *args):
        """
        Initializes an Isomer anonymous user interface event.

        :param action:
        :param data:
        :param client:
        :param args:
        :return:
        """

        self.name = self.__module__ + "." + self.__class__.__name__
        super(anonymous_event, self).__init__(*args)
        self.action = action
        self.data = data
        self.client = client
        isolog("AnonymousEvent created:", self.name, lvl=events)

    @classmethod
    def realname(cls):
        """Return real name of an object class"""

        # For circuits manager to enable module/event namespaces
        return cls.__module__ + "." + cls.__name__


class authorized_event(isomer_ui_event):
    """Base class for events for logged in users."""

    roles = ["admin", "crew"]

    def __init__(self, user, action, data, client, *args):
        """
        Initializes an Isomer authorized user interface event.

        :param user: User object from :py:class:isomer.web.clientmanager.User
        :param action:
        :param data:
        :param client:
        :param args:
        :return:
        """

        # assert isinstance(user, User)

        self.name = self.__module__ + "." + self.__class__.__name__
        super(authorized_event, self).__init__(*args)
        self.user = user
        self.action = action
        self.data = data
        self.client = client
        isolog("AuthorizedEvent created:", self.name, lvl=events)

    @classmethod
    def realname(cls):
        """Return real name of an object class"""

        # For circuits manager to enable module/event namespaces
        return cls.__module__ + "." + cls.__name__

    @classmethod
    def source(cls):
        """Return real name of an object class"""

        # For circuits manager to enable module/event namespaces
        return cls.__module__


class system_stop(isomer_event):
    """Stop everything, save persistent state and cease operations"""


# Configuration reload event

# TODO: This should probably not be an ui-event
class reload_configuration(isomer_ui_event):
    """Instructs a component to reload its configuration"""

    def __init__(self, target, *args, **kwargs):
        super(reload_configuration, self).__init__(*args, **kwargs)
        self.target = target
        isolog("Reload of configuration triggered", lvl=events)


# Authenticator Events


class profilerequest(authorized_event):
    """A user has changed his profile"""

    def __init__(self, *args):
        """

        :param user: Userobject of client
        :param data: The new profile data
        """
        super(profilerequest, self).__init__(*args)

        isolog(
            "Profile update request: ",
            self.__dict__,
            lvl=events,
            emitter="PROFILE-EVENT",
        )


# Frontend assembly events


class frontendbuildrequest(Event):
    """Rebuild and/or install the frontend"""

    def __init__(self, force=False, install=False, *args):
        super(frontendbuildrequest, self).__init__(*args)
        self.force = force
        self.install = install


class componentupdaterequest(frontendbuildrequest):
    """Check for updated components"""

    pass


# Debugger


class logtailrequest(authorized_event):
    """Request the logger's latest output"""

    pass


class debugrequest(authorized_event):
    """Debugging event"""

    def __init__(self, *args):
        super(debugrequest, self).__init__(*args)

        isolog("Created debugrequest", lvl=events, emitter="DEBUG-EVENT")


asyncapi_template = {
    "asyncapi": "2.0.0",
    "info": {
        "title": "isomer",
        "version": "2.0.0",
        "contact": {
            "name": "Isomer API Support",
            "url": "http://github.com/isomeric/api",
            "email": "[email protected]"
        },
        "license": {
            "name": "AGPL 3.0",
            "url": "http://www.gnu.org/licenses/agpl-3.0.en.html"
        },
        "description": "This is a local Isomer API.",

    },
    "tags": [
        {
            "name": "isomer",
            "description": "Isomer Application Framework"
        }
    ],
    "servers": {
        "development": {
            "url": "ws://localhost:15674/ws",
            "description": "Local rabbitmq server with stomp",
            "protocol": "stomp",
            "protocolVersion": "1.2.0"
        },
        "docker-compose": {
            "url": "ws://rabbit_container:15674/ws",
            "description": "Local docker composer based rabbitmq server with stomp",
            "protocol": "stomp",
            "protocolVersion": "1.2.0"
        }
    },
    "channels": {}
}


# {
#    'event': <class 'isomer.ui.configurator.get'>,
#    'name': 'isomer.ui.configurator.get',
#    'doc': 'A client requires a schema to validate data or display a form',
#    'args': {'uuid': jsonschema}
# }

# {'uuid': {'description': 'Select an object',
#           'pattern': '^[a-fA-F0-9]*$',
#           'title': 'Reference',
#           'type': 'string'}}

def generate_asyncapi():
    """Generate async-api definition"""

    if not populated:
        populate_user_events()

    api_events = {**AuthorizedEvents, **AnonymousEvents}

    api = copy(asyncapi_template)

    for package, channel_events in api_events.items():
        isolog('Inspecting package:', package)
        for name, meta in channel_events.items():
            isolog(meta, lvl=verbose)

            if meta['args'] == {}:
                isolog(name.ljust(20), ":", meta, pretty=True, lvl=debug)

            else:
                isolog(meta['args'], pretty=True, lvl=debug)
                channel, event_name = meta['name'].rsplit('.', 1)
                channel = channel.replace(".", "/")

                if channel not in api['channels']:
                    api['channels'][channel] = {}

                api['channels'][channel][event_name] = {
                    "summary": meta["summary"],
                    "tags": meta["tags"],
                    "description": meta["doc"],
                    'operationId': event_name,
                    "message": {
                        "payload": meta['args'],
                    }
                }

    isolog("\n", api, pretty=True, lvl=hilight)


    return api




1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			# Isomer - The distributed application framework
5			# ==============================================
6			# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
7			#
8			# This program is free software: you can redistribute it and/or modify
9			# it under the terms of the GNU Affero General Public License as published by
10			# the Free Software Foundation, either version 3 of the License, or
11			# (at your option) any later version.
12			#
13			# This program is distributed in the hope that it will be useful,
14			# but WITHOUT ANY WARRANTY; without even the implied warranty of
15			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16			# GNU Affero General Public License for more details.
17			#
18			# You should have received a copy of the GNU Affero General Public License
19			# along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21			"""
22
23
24			Module: Events
25			==============
26
27			Major Isomer event declarations
28
29
30			"""
31
32			from copy import copy
33			from typing import Dict
34
35			from circuits.core import Event
36			from isomer.logger import isolog, events
37
38			# from isomer.ui.clientobjects import User
39
40
41			AuthorizedEvents: Dict[str, Event] = {}
42			AnonymousEvents: Dict[str, Event] = {}
43
44			populated = False
45
46
47			def get_user_events():
48			"""Return all registered authorized events"""
49
50			return AuthorizedEvents
51
52
53			def get_anonymous_events():
54			"""Return all registered anonymous events"""
55
56			return AnonymousEvents
57
58
59			def populate_user_events():
60			"""Generate a list of all registered authorized and anonymous events"""
61
62			global AuthorizedEvents
63			global AnonymousEvents
64			global populated
65
66			def inheritors(klass):
67			"""Find inheritors of a specified object class"""
68
69			subclasses = {}
70			subclasses_set = set()
71			work = [klass]
72			while work:
73			parent = work.pop()
74			for child in parent.__subclasses__():
75			if child not in subclasses_set:
76			# pprint(child.__dict__)
77			name = child.__module__ + "." + child.__name__
78
79			subclasses_set.add(child)
80			event = {
81			"event": child,
82			"name": name,
83			"doc": child.__doc__,
84			"summary": child.summary,
85			"tags": child.tags,
86			"args": child.args,
87			}
88
89			if child.__module__ in subclasses:
90			subclasses[child.__module__][child.__name__] = event
91			else:
92			subclasses[child.__module__] = {child.__name__: event}
93			work.append(child)
94			return subclasses
95
96			# TODO: Change event system again, to catch authorized (i.e. "user") as
97			# well as normal events, so they can be processed by Automat
98
99			# NormalEvents = inheritors(Event)
100			AuthorizedEvents = inheritors(authorized_event)
101			AnonymousEvents = inheritors(anonymous_event)
102
103			# AuthorizedEvents.update(NormalEvents)
104			populated = True
105
106
107			class isomer_basic_event(Event):
108			"""Basic Isomer event class"""
109
110			args = {}
111			tags = []
112			summary = "Basic Isomer Event"
113
114			def __init__(self, args, *kwargs):
115			"""Initializes a basic Isomer event.
116
117			For further details, check out the circuits documentation.
118			"""
119			super(isomer_basic_event, self).__init__(args, *kwargs)
120
121
122			class isomer_ui_event(isomer_basic_event):
123			"""Isomer user interface event class"""
124
125			pass
126
127
128			class isomer_event(isomer_basic_event):
129			"""Isomer internal event class"""
130
131			pass
132
133
134			class anonymous_event(isomer_ui_event):
135			"""Base class for events for logged in users."""
136
137			def __init__(self, action, data, client, *args):
138			"""
139			Initializes an Isomer anonymous user interface event.
140
141			:param action:
142			:param data:
143			:param client:
144			:param args:
145			:return:
146			"""
147
148			self.name = self.__module__ + "." + self.__class__.__name__
149			super(anonymous_event, self).__init__(*args)
150			self.action = action
151			self.data = data
152			self.client = client
153			isolog("AnonymousEvent created:", self.name, lvl=events)
154
155			@classmethod
156			def realname(cls):
157			"""Return real name of an object class"""
158
159			# For circuits manager to enable module/event namespaces
160			return cls.__module__ + "." + cls.__name__
161
162
163			class authorized_event(isomer_ui_event):
164			"""Base class for events for logged in users."""
165
166			roles = ["admin", "crew"]
167
168			def __init__(self, user, action, data, client, *args):
169			"""
170			Initializes an Isomer authorized user interface event.
171
172			:param user: User object from :py:class:isomer.web.clientmanager.User
173			:param action:
174			:param data:
175			:param client:
176			:param args:
177			:return:
178			"""
179
180			# assert isinstance(user, User)
181
182			self.name = self.__module__ + "." + self.__class__.__name__
183			super(authorized_event, self).__init__(*args)
184			self.user = user
185			self.action = action
186			self.data = data
187			self.client = client
188			isolog("AuthorizedEvent created:", self.name, lvl=events)
189
190			@classmethod
191			def realname(cls):
192			"""Return real name of an object class"""
193
194			# For circuits manager to enable module/event namespaces
195			return cls.__module__ + "." + cls.__name__
196
197			@classmethod
198			def source(cls):
199			"""Return real name of an object class"""
200
201			# For circuits manager to enable module/event namespaces
202			return cls.__module__
203
204
205			class system_stop(isomer_event):
206			"""Stop everything, save persistent state and cease operations"""
207
208
209			# Configuration reload event
210
211			# TODO: This should probably not be an ui-event
212			class reload_configuration(isomer_ui_event):
213			"""Instructs a component to reload its configuration"""
214
215			def __init__(self, target, args, *kwargs):
216			super(reload_configuration, self).__init__(args, *kwargs)
217			self.target = target
218			isolog("Reload of configuration triggered", lvl=events)
219
220
221			# Authenticator Events
222
223
224			class profilerequest(authorized_event):
225			"""A user has changed his profile"""
226
227			def __init__(self, *args):
228			"""
229
230			:param user: Userobject of client
231			:param data: The new profile data
232			"""
233			super(profilerequest, self).__init__(*args)
234
235			isolog(
236			"Profile update request: ",
237			self.__dict__,
238			lvl=events,
239			emitter="PROFILE-EVENT",
240			)
241
242
243			# Frontend assembly events
244
245
246			class frontendbuildrequest(Event):
247			"""Rebuild and/or install the frontend"""
248
249			def __init__(self, force=False, install=False, *args):
250			super(frontendbuildrequest, self).__init__(*args)
251			self.force = force
252			self.install = install
253
254
255			class componentupdaterequest(frontendbuildrequest):
256			"""Check for updated components"""
257
258			pass
259
260
261			# Debugger
262
263
264			class logtailrequest(authorized_event):
265			"""Request the logger's latest output"""
266
267			pass
268
269
270			class debugrequest(authorized_event):
271			"""Debugging event"""
272
273			def __init__(self, *args):
274			super(debugrequest, self).__init__(*args)
275
276			isolog("Created debugrequest", lvl=events, emitter="DEBUG-EVENT")
277
278
279			asyncapi_template = {
280			"asyncapi": "2.0.0",
281			"info": {
282			"title": "isomer",
283			"version": "2.0.0",
284			"contact": {
285			"name": "Isomer API Support",
286			"url": "http://github.com/isomeric/api",
287			"email": "[email protected]"
288			},
289			"license": {
290			"name": "AGPL 3.0",
291			"url": "http://www.gnu.org/licenses/agpl-3.0.en.html"
292			},
293			"description": "This is a local Isomer API.",
294
295			},
296			"tags": [
297			{
298			"name": "isomer",
299			"description": "Isomer Application Framework"
300			}
301			],
302			"servers": {
303			"development": {
304			"url": "ws://localhost:15674/ws",
305			"description": "Local rabbitmq server with stomp",
306			"protocol": "stomp",
307			"protocolVersion": "1.2.0"
308			},
309			"docker-compose": {
310			"url": "ws://rabbit_container:15674/ws",
311			"description": "Local docker composer based rabbitmq server with stomp",
312			"protocol": "stomp",
313			"protocolVersion": "1.2.0"
314			}
315			},
316			"channels": {}
317			}
318
319
320			# {
321			# 'event': <class 'isomer.ui.configurator.get'>,
322			# 'name': 'isomer.ui.configurator.get',
323			# 'doc': 'A client requires a schema to validate data or display a form',
324			# 'args': {'uuid': jsonschema}
325			# }
326
327			# {'uuid': {'description': 'Select an object',
328			# 'pattern': '^[a-fA-F0-9]*$',
329			# 'title': 'Reference',
330			# 'type': 'string'}}
331
332			def generate_asyncapi():
333			"""Generate async-api definition"""
334
335			if not populated:
336			populate_user_events()
337
338			api_events = {AuthorizedEvents, AnonymousEvents}
339
340			api = copy(asyncapi_template)
341
342			for package, channel_events in api_events.items():
343			isolog('Inspecting package:', package)
344			for name, meta in channel_events.items():
345			isolog(meta, lvl=verbose)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-18 04:12 UTC by Report Bug Copy Issue Report The variable `verbose` does not seem to be defined. Loading history...
346			if meta['args'] == {}:
347			isolog(name.ljust(20), ":", meta, pretty=True, lvl=debug)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-18 04:12 UTC by Report Bug Copy Issue Report The variable `debug` does not seem to be defined. Loading history...
348			else:
349			isolog(meta['args'], pretty=True, lvl=debug)
350			channel, event_name = meta['name'].rsplit('.', 1)
351			channel = channel.replace(".", "/")
352
353			if channel not in api['channels']:
354			api['channels'][channel] = {}
355
356			api['channels'][channel][event_name] = {
357			"summary": meta["summary"],
358			"tags": meta["tags"],
359			"description": meta["doc"],
360			'operationId': event_name,
361			"message": {
362			"payload": meta['args'],
363			}
364			}
365
366			isolog("\n", api, pretty=True, lvl=hilight)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-12-18 04:12 UTC by Report Bug Copy Issue Report The variable `hilight` does not seem to be defined. Loading history...
367
368			return api
369
370
371

isomeric / isomer

Push — master ( ef441d...0acd10 )

isomer.events.system.generate_asyncapi() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like