kytos.core.napps.base.NApp.uri() - Code Metrics - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

kytos.core.napps.base.NApp.uri() A
last analyzed 2025-06-18 19:57 UTC

↳ Parent: kytos.core.napps.base

Complexity

Conditions

Size

Total Lines	12
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	1
dl	0
loc	12
rs	10
c	0
b	0
f	0

"""Kytos Napps Module."""
import json
import os
import re
import sys
import tarfile
import traceback
import urllib
from abc import ABCMeta, abstractmethod
from pathlib import Path
from random import randint
from threading import Event, Thread

# pylint: disable=consider-using-from-import,redefined-outer-name
import kytos.core.controller as controller
from kytos.core.events import KytosEvent
from kytos.core.logs import NAppLog

__all__ = ('KytosNApp',)

LOG = NAppLog()


class NApp:
    """Class to represent a NApp."""

    # pylint: disable=too-many-arguments
    def __init__(self, username=None, name=None, version=None,
                 repository=None, meta=False):
        self.username = username
        self.name = name
        self.version = version if version else 'latest'
        self.repository = repository
        self.meta = meta
        self.description = None
        self.tags = []
        self.enabled = False
        self.napp_dependencies = []

    def __str__(self):
        return f"{self.username}/{self.name}"

    def __repr__(self):
        return f"NApp({self.username}/{self.name})"

    def __hash__(self):
        return hash(self.id)

    def __eq__(self, other):
        """Compare username/name strings."""
        return isinstance(other, self.__class__) and self.id == other.id

    @property
    def id(self):  # pylint: disable=invalid-name
        """username/name string."""
        return str(self)

    @property
    def uri(self):
        """Return a unique identifier of this NApp."""
        version = self.version if self.version else 'latest'

        if not self._has_valid_repository():
            return ""

        # Use the next line after Diraol fix redirect using ":" for version
        # return "{}/{}:{}".format(self.repository, self.id, version)

        return f"{self.repository}/{self.id}-{version}"

    @property
    def package_url(self):
        """Return a fully qualified URL for a NApp package."""
        if not self.uri:
            return ""
        return f"{self.uri}.napp"

    @classmethod
    def create_from_uri(cls, uri):
        """Return a new NApp instance from an unique identifier."""
        regex = r'^(((https?://|file://)(.+))/)?(.+?)/(.+?)/?(:(.+))?$'
        match = re.match(regex, uri)

        if not match:
            return None

        return cls(username=match.groups()[4],
                   name=match.groups()[5],
                   version=match.groups()[7],
                   repository=match.groups()[1])

    @classmethod
    def create_from_json(cls, filename):
        """Return a new NApp instance from a metadata file."""
        with open(filename, encoding='utf-8') as data_file:
            data = json.loads(data_file.read())

        return cls.create_from_dict(data)

    @classmethod
    def create_from_dict(cls, data):
        """Return a new NApp instance from metadata."""
        napp = cls()

        for attribute, value in data.items():
            setattr(napp, attribute, value)

        return napp

    def as_json(self):
        """Dump all NApp attributes on a json format."""
        return json.dumps(self.__dict__)

    def match(self, pattern):
        """Whether a pattern is present on NApp id, description and tags."""
        try:
            pattern = f'.*{pattern}.*'
            pattern = re.compile(pattern, re.IGNORECASE)
            strings = [self.id, self.description] + self.tags
            return any(pattern.match(string) for string in strings)
        except TypeError:
            return False

    def download(self):
        """Download NApp package from his repository.

        Raises:
            urllib.error.HTTPError: If download is not successful.

        Returns:
            str: Downloaded temp filename.

        """
        if not self.package_url:
            return None

        package_filename = urllib.request.urlretrieve(self.package_url)[0]
        extracted = self._extract(package_filename)
        Path(package_filename).unlink()
        self._update_repo_file(extracted)
        return extracted

    @staticmethod
    def _extract(filename):
        """Extract NApp package to a temporary folder.

        Return:
            pathlib.Path: Temp dir with package contents.
        """
        random_string = str(randint(0, 10**6))
        tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
        os.mkdir(tmp)
        with tarfile.open(filename, 'r:xz') as tar:
            tar.extractall(tmp)
        return Path(tmp)

    def _has_valid_repository(self):
        """Whether this NApp has a valid repository or not."""
        return all([self.username, self.name, self.repository])

    def _update_repo_file(self, destination=None):
        """Create or update the file '.repo' inside NApp package."""
        with open(f"{destination}/.repo", 'w', encoding="utf8") as repo_file:
            repo_file.write(self.repository + '\n')


class KytosNApp(Thread, metaclass=ABCMeta):
    """Base class for any KytosNApp to be developed."""

    def __init__(self, controller: "controller.Controller", **kwargs):
        """Contructor of KytosNapps.

        Go through all of the instance methods and selects those that have
        the events attribute, then creates a dict containing the event_name
        and the list of methods that are responsible for handling such event.

        At the end, the setup method is called as a complement of the init
        process.
        """
        Thread.__init__(self, daemon=False)
        self.controller = controller
        self.username = None  # loaded from json
        self.name = None      # loaded from json
        self.meta = False     # loaded from json
        self._load_json()

        # Force a listener with a private method.
        self._listeners = {
            'kytos/core.shutdown': [self._shutdown_handler],
            'kytos/core.shutdown.' + self.napp_id: [self._shutdown_handler]}

        self.__event = Event()
        #: int: Seconds to sleep before next call to :meth:`execute`. If
        #: negative, run :meth:`execute` only once.
        self.__interval = -1
        self.setup()

        #: Add non-private methods that listen to events.
        handler_methods = [getattr(self, method_name) for method_name in
                           dir(self) if method_name[0] != '_' and
                           callable(getattr(self, method_name)) and
                           hasattr(getattr(self, method_name), 'events')]

        # Building the listeners dictionary
        for method in handler_methods:
            for event_name in method.events:
                if event_name not in self._listeners:
                    self._listeners[event_name] = []
                self._listeners[event_name].append(method)

    @property
    def napp_id(self):
        """username/name string."""
        return f"{self.username}/{self.name}"

    def listeners(self):
        """Return all listeners registered."""
        return list(self._listeners.keys())

    def _load_json(self):
        """Update object attributes based on kytos.json."""
        current_file = sys.modules[self.__class__.__module__].__file__
        json_path = os.path.join(os.path.dirname(current_file), 'kytos.json')
        with open(json_path, encoding='utf-8') as data_file:
            data = json.loads(data_file.read())

        for attribute, value in data.items():
            setattr(self, attribute, value)

    def execute_as_loop(self, interval):
        """Run :meth:`execute` within a loop. Call this method during setup.

        By calling this method, the application does not need to worry about
        loop details such as sleeping and stopping the loop when
        :meth:`shutdown` is called. Just call this method during :meth:`setup`
        and implement :meth:`execute` as a single execution.

        Args:
            interval (int): Seconds between each call to :meth:`execute`.
        """
        self.__interval = interval

    # pylint: disable=broad-except
    def run(self):
        """Call the execute method, looping as needed.

        It should not be overriden.
        """
        self.notify_loaded()
        LOG.info(f"Running NApp: {self}")
        try:
            self.execute()
        except Exception:
            traceback_str = traceback.format_exc(chain=False)
            LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
        while self.__interval > 0:
            self.__event.wait(self.__interval)
            if self.__event.is_set():
                break
            try:
                self.execute()
            except Exception:
                traceback_str = traceback.format_exc(chain=False)
                LOG.error(f"NApp: {self} unhandled exception {traceback_str}")

    def notify_loaded(self):
        """Inform this NApp has been loaded."""
        name = f'{self.username}/{self.name}.loaded'
        event = KytosEvent(name=name, content={})
        self.controller.buffers.meta.put(event)

    # all listeners receive event
    def _shutdown_handler(self, event):  # pylint: disable=unused-argument
        """Listen shutdown event from kytos.

        This method listens the kytos/core.shutdown event and call the shutdown
        method from napp subclass implementation.

        Paramters
            event (:class:`KytosEvent`): event to be listened.
        """
        if not self.__event.is_set():
            self.shutdown()
            self.__event.set()

    @abstractmethod
    def setup(self):
        """Replace the 'init' method for the KytosApp subclass.

        The setup method is automatically called on the NApp __init__().
        Users aren't supposed to call this method directly.
        """

    @abstractmethod
    def execute(self):
        """Execute in a loop until 'kytos/core.shutdown' is received.

        The execute method is called by KytosNApp class.
        Users shouldn't call this method directly.
        """

    @abstractmethod
    def shutdown(self):
        """Run before the app is unloaded and the controller, stopped.

        The user implementation of this method should do the necessary routine
        for the user App and also it is a good moment to break the loop of the
        execute method if it is in a loop.

        This methods is not going to be called directly, it is going to be
        called by the _shutdown_handler method when a KytosShutdownEvent is
        sent.
        """


1			"""Kytos Napps Module."""
2			import json
3			import os
4			import re
5			import sys
6			import tarfile
7			import traceback
8			import urllib
9			from abc import ABCMeta, abstractmethod
10			from pathlib import Path
11			from random import randint
12			from threading import Event, Thread
13
14			# pylint: disable=consider-using-from-import,redefined-outer-name
15			import kytos.core.controller as controller
16			from kytos.core.events import KytosEvent
17			from kytos.core.logs import NAppLog
18
19			__all__ = ('KytosNApp',)
20
21			LOG = NAppLog()
22
23
24			class NApp:
25			"""Class to represent a NApp."""
26
27			# pylint: disable=too-many-arguments
28			def __init__(self, username=None, name=None, version=None,
29			repository=None, meta=False):
30			self.username = username
31			self.name = name
32			self.version = version if version else 'latest'
33			self.repository = repository
34			self.meta = meta
35			self.description = None
36			self.tags = []
37			self.enabled = False
38			self.napp_dependencies = []
39
40			def __str__(self):
41			return f"{self.username}/{self.name}"
42
43			def __repr__(self):
44			return f"NApp({self.username}/{self.name})"
45
46			def __hash__(self):
47			return hash(self.id)
48
49			def __eq__(self, other):
50			"""Compare username/name strings."""
51			return isinstance(other, self.__class__) and self.id == other.id
52
53			@property
54			def id(self): # pylint: disable=invalid-name
55			"""username/name string."""
56			return str(self)
57
58			@property
59			def uri(self):
60			"""Return a unique identifier of this NApp."""
61			version = self.version if self.version else 'latest'
62
63			if not self._has_valid_repository():
64			return ""
65
66			# Use the next line after Diraol fix redirect using ":" for version
67			# return "{}/{}:{}".format(self.repository, self.id, version)
68
69			return f"{self.repository}/{self.id}-{version}"
70
71			@property
72			def package_url(self):
73			"""Return a fully qualified URL for a NApp package."""
74			if not self.uri:
75			return ""
76			return f"{self.uri}.napp"
77
78			@classmethod
79			def create_from_uri(cls, uri):
80			"""Return a new NApp instance from an unique identifier."""
81			regex = r'^(((https?://\|file://)(.+))/)?(.+?)/(.+?)/?(:(.+))?$'
82			match = re.match(regex, uri)
83
84			if not match:
85			return None
86
87			return cls(username=match.groups()[4],
88			name=match.groups()[5],
89			version=match.groups()[7],
90			repository=match.groups()[1])
91
92			@classmethod
93			def create_from_json(cls, filename):
94			"""Return a new NApp instance from a metadata file."""
95			with open(filename, encoding='utf-8') as data_file:
96			data = json.loads(data_file.read())
97
98			return cls.create_from_dict(data)
99
100			@classmethod
101			def create_from_dict(cls, data):
102			"""Return a new NApp instance from metadata."""
103			napp = cls()
104
105			for attribute, value in data.items():
106			setattr(napp, attribute, value)
107
108			return napp
109
110			def as_json(self):
111			"""Dump all NApp attributes on a json format."""
112			return json.dumps(self.__dict__)
113
114			def match(self, pattern):
115			"""Whether a pattern is present on NApp id, description and tags."""
116			try:
117			pattern = f'.{pattern}.'
118			pattern = re.compile(pattern, re.IGNORECASE)
119			strings = [self.id, self.description] + self.tags
120			return any(pattern.match(string) for string in strings)
121			except TypeError:
122			return False
123
124			def download(self):
125			"""Download NApp package from his repository.
126
127			Raises:
128			urllib.error.HTTPError: If download is not successful.
129
130			Returns:
131			str: Downloaded temp filename.
132
133			"""
134			if not self.package_url:
135			return None
136
137			package_filename = urllib.request.urlretrieve(self.package_url)[0]
138			extracted = self._extract(package_filename)
139			Path(package_filename).unlink()
140			self._update_repo_file(extracted)
141			return extracted
142
143			@staticmethod
144			def _extract(filename):
145			"""Extract NApp package to a temporary folder.
146
147			Return:
148			pathlib.Path: Temp dir with package contents.
149			"""
150			random_string = str(randint(0, 10**6))
151			tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
152			os.mkdir(tmp)
153			with tarfile.open(filename, 'r:xz') as tar:
154			tar.extractall(tmp)
155			return Path(tmp)
156
157			def _has_valid_repository(self):
158			"""Whether this NApp has a valid repository or not."""
159			return all([self.username, self.name, self.repository])
160
161			def _update_repo_file(self, destination=None):
162			"""Create or update the file '.repo' inside NApp package."""
163			with open(f"{destination}/.repo", 'w', encoding="utf8") as repo_file:
164			repo_file.write(self.repository + '\n')
165
166
167			class KytosNApp(Thread, metaclass=ABCMeta):
168			"""Base class for any KytosNApp to be developed."""
169
170			def __init__(self, controller: "controller.Controller", **kwargs):
171			"""Contructor of KytosNapps.
172
173			Go through all of the instance methods and selects those that have
174			the events attribute, then creates a dict containing the event_name
175			and the list of methods that are responsible for handling such event.
176
177			At the end, the setup method is called as a complement of the init
178			process.
179			"""
180			Thread.__init__(self, daemon=False)
181			self.controller = controller
182			self.username = None # loaded from json
183			self.name = None # loaded from json
184			self.meta = False # loaded from json
185			self._load_json()
186
187			# Force a listener with a private method.
188			self._listeners = {
189			'kytos/core.shutdown': [self._shutdown_handler],
190			'kytos/core.shutdown.' + self.napp_id: [self._shutdown_handler]}
191
192			self.__event = Event()
193			#: int: Seconds to sleep before next call to :meth:`execute`. If
194			#: negative, run :meth:`execute` only once.
195			self.__interval = -1
196			self.setup()
197
198			#: Add non-private methods that listen to events.
199			handler_methods = [getattr(self, method_name) for method_name in
200			dir(self) if method_name[0] != '_' and
201			callable(getattr(self, method_name)) and
202			hasattr(getattr(self, method_name), 'events')]
203
204			# Building the listeners dictionary
205			for method in handler_methods:
206			for event_name in method.events:
207			if event_name not in self._listeners:
208			self._listeners[event_name] = []
209			self._listeners[event_name].append(method)
210
211			@property
212			def napp_id(self):
213			"""username/name string."""
214			return f"{self.username}/{self.name}"
215
216			def listeners(self):
217			"""Return all listeners registered."""
218			return list(self._listeners.keys())
219
220			def _load_json(self):
221			"""Update object attributes based on kytos.json."""
222			current_file = sys.modules[self.__class__.__module__].__file__
223			json_path = os.path.join(os.path.dirname(current_file), 'kytos.json')
224			with open(json_path, encoding='utf-8') as data_file:
225			data = json.loads(data_file.read())
226
227			for attribute, value in data.items():
228			setattr(self, attribute, value)
229
230			def execute_as_loop(self, interval):
231			"""Run :meth:`execute` within a loop. Call this method during setup.
232
233			By calling this method, the application does not need to worry about
234			loop details such as sleeping and stopping the loop when
235			:meth:`shutdown` is called. Just call this method during :meth:`setup`
236			and implement :meth:`execute` as a single execution.
237
238			Args:
239			interval (int): Seconds between each call to :meth:`execute`.
240			"""
241			self.__interval = interval
242
243			# pylint: disable=broad-except
244			def run(self):
245			"""Call the execute method, looping as needed.
246
247			It should not be overriden.
248			"""
249			self.notify_loaded()
250			LOG.info(f"Running NApp: {self}")
251			try:
252			self.execute()
253			except Exception:
254			traceback_str = traceback.format_exc(chain=False)
255			LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
256			while self.__interval > 0:
257			self.__event.wait(self.__interval)
258			if self.__event.is_set():
259			break
260			try:
261			self.execute()
262			except Exception:
263			traceback_str = traceback.format_exc(chain=False)
264			LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
265
266			def notify_loaded(self):
267			"""Inform this NApp has been loaded."""
268			name = f'{self.username}/{self.name}.loaded'
269			event = KytosEvent(name=name, content={})
270			self.controller.buffers.meta.put(event)
271
272			# all listeners receive event
273			def _shutdown_handler(self, event): # pylint: disable=unused-argument
274			"""Listen shutdown event from kytos.
275
276			This method listens the kytos/core.shutdown event and call the shutdown
277			method from napp subclass implementation.
278
279			Paramters
280			event (:class:`KytosEvent`): event to be listened.
281			"""
282			if not self.__event.is_set():
283			self.shutdown()
284			self.__event.set()
285
286			@abstractmethod
287			def setup(self):
288			"""Replace the 'init' method for the KytosApp subclass.
289
290			The setup method is automatically called on the NApp __init__().
291			Users aren't supposed to call this method directly.
292			"""
293
294			@abstractmethod
295			def execute(self):
296			"""Execute in a loop until 'kytos/core.shutdown' is received.
297
298			The execute method is called by KytosNApp class.
299			Users shouldn't call this method directly.
300			"""
301
302			@abstractmethod
303			def shutdown(self):
304			"""Run before the app is unloaded and the controller, stopped.
305
306			The user implementation of this method should do the necessary routine
307			for the user App and also it is a good moment to break the loop of the
308			execute method if it is in a loop.
309
310			This methods is not going to be called directly, it is going to be
311			called by the _shutdown_handler method when a KytosShutdownEvent is
312			sent.
313			"""
314

kytos-ng / kytos

kytos.core.napps.base.NApp.uri() A last analyzed 2025-06-18 19:57 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.napps.base.NApp.uri() A
last analyzed 2025-06-18 19:57 UTC