NAppsAPI.list() - Code Metrics - Inspection of "Fix URLError when installing napp from remote" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#110)

by macartur

created 2017-07-24 17:45 UTC

NAppsAPI.list() A

↳ Parent: NAppsAPI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	4
c	3
b	1
f	0
dl	0
loc	17
rs	9.2

"""Translate cli commands to non-cli code."""
import json
import logging
import os
import re
from urllib.error import HTTPError, URLError

import requests

from kytos.utils.napps import NAppsManager

log = logging.getLogger(__name__)


class NAppsAPI:
    """An API for the command-line interface.

    Use the config file only for required options. Static methods are called
    by the parser and they instantiate an object of this class to fulfill the
    request.
    """

    @classmethod
    def disable(cls, args):
        """Disable subcommand."""
        mgr = NAppsManager()

        if args['all']:
            napps = mgr.get_enabled()
        else:
            napps = args['<napp>']

        for napp in napps:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            cls.disable_napp(mgr)

    @staticmethod
    def disable_napp(mgr):
        """Disable a NApp."""
        if mgr.is_enabled():
            log.info('  Disabling...')
            mgr.disable()
        log.info('  Disabled.')

    @classmethod
    def enable(cls, args):
        """Enable subcommand."""
        mgr = NAppsManager()

        if args['all']:
            napps = mgr.get_disabled()
        else:
            napps = args['<napp>']

        cls.enable_napps(napps)

    @classmethod
    def enable_napp(cls, mgr):
        """Install one NApp using NAppManager object."""
        try:
            if not mgr.is_enabled():
                log.info('  Enabling...')
                mgr.enable()
            log.info('  Enabled.')
            cls.enable_napps(mgr.dependencies())
        except (FileNotFoundError, PermissionError) as e:
            log.error('  %s', e)

    @classmethod
    def enable_napps(cls, napps):
        """Enable a list of NApps.

        Args:
            napps (list): List of NApps.
        """
        mgr = NAppsManager()
        for napp in napps:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            cls.enable_napp(mgr)

    @classmethod
    def create(cls, args):
        """Bootstrap a basic NApp structure on the current folder."""
        NAppsManager.create_napp()

    @classmethod
    def upload(cls, args):
        """Upload the NApp to the NApps server.

        Create the NApp package and upload it to the NApp server.
        """
        try:
            NAppsManager().upload()
        except FileNotFoundError:
            log.error("Couldn't find kytos.json in current directory.")

    @classmethod
    def uninstall(cls, args):
        """Uninstall and delete NApps.

        For local installations, do not delete code outside install_path and
        enabled_path.
        """
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            if mgr.is_installed():
                if mgr.is_enabled():
                    cls.disable_napp(mgr)
                log.info('  Uninstalling...')
                mgr.uninstall()
            log.info('  Uninstalled.')

    @classmethod
    def install(cls, args):
        """Install local or remote NApps."""
        cls.install_napps(args['<napp>'])

    @classmethod
    def install_napps(cls, napps):
        """Install local or remote NApps.

        This method is recursive, it will install each napps and your
        dependencies.
        """
        mgr = NAppsManager()
        for napp in napps:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            if not mgr.is_installed():
                cls.install_napp(mgr)
            else:
                log.info('  Installed.')
                cls.enable_napp(mgr)
            napp_dependencies = mgr.dependencies()
            if napp_dependencies:
                log.info('Installing Dependencies:')
                cls.install_napps(napp_dependencies)

    @classmethod
    def install_napp(cls, mgr):
        """Install a NApp."""
        try:
            log.info('  Searching local NApp...')
            mgr.install_local()
            log.info('  Found and installed.')
            cls.enable_napp(mgr)
        except FileNotFoundError:
            log.info('  Not found. Downloading from NApps Server...')
            try:
                mgr.install_remote()
                log.info('  Downloaded and installed.')
                cls.enable_napp(mgr)
            except HTTPError as e:
                if e.code == 404:
                    log.error('  NApp not found.')
                else:
                    log.error('  NApps Server error: %s', e)
            except URLError as e:
                log.error('  NApps Server error: '+str(e.reason))

    @classmethod
    def search(cls, args):
        """Search for NApps in NApps server matching a pattern."""
        safe_shell_pat = re.escape(args['<pattern>']).replace(r'\*', '.*')
        pat_str = '.*{}.*'.format(safe_shell_pat)
        pattern = re.compile(pat_str, re.IGNORECASE)
        remote_json = NAppsManager.search(pattern)
        remote = set()
        for napp in remote_json:
            # WARNING: This will be changed in future versions, when 'author'
            # will be removed.
            username = napp.get('username', napp.get('author'))
            remote.add(((username, napp.get('name')), napp.get('description')))

        cls._print_napps(remote)

    @classmethod
    def _print_napps(cls, napp_list):
        """Format the NApp list to be printed."""
        mgr = NAppsManager()
        enabled = mgr.get_enabled()
        installed = mgr.get_installed()
        napps = []
        for napp, desc in sorted(napp_list):
            status = 'i' if napp in installed else '-'
            status += 'e' if napp in enabled else '-'
            status = '[{}]'.format(status)
            name = '{}/{}'.format(*napp)
            napps.append((status, name, desc))
        cls.print_napps(napps)

    @classmethod
    def list(cls, args):
        """List all installed NApps and inform whether they are enabled."""
        mgr = NAppsManager()

        # Add status
        napps = [napp + ('[ie]',) for napp in mgr.get_enabled()]
        napps += [napp + ('[i-]',) for napp in mgr.get_disabled()]

        # Sort, add description and reorder coloumns
        napps.sort()
        napps_ordered = []
        for user, name, status in napps:
            napps_ordered.append((status, '{}/{}'.format(user, name),
                                  mgr.get_description(user, name)))

        cls.print_napps(napps_ordered)

    @staticmethod
    def print_napps(napps):
        """Print status, name and description."""
        if not napps:
            print('No NApps found.')
            return

        stat_w = 6  # We already know the size of Status col
        name_w = max(len(n[1]) for n in napps)
        desc_w = max(len(n[2]) for n in napps)
        term_w = os.popen('stty size', 'r').read().split()[1]
        remaining = int(term_w) - stat_w - name_w - 6
        desc_w = min(desc_w, remaining)
        widths = (stat_w, name_w, desc_w)

        header = '\n{:^%d} | {:^%d} | {:^%d}' % widths
        row = '{:^%d} | {:<%d} | {:<%d}' % widths
        print(header.format('Status', 'NApp ID', 'Description'))
        print('=+='.join('=' * w for w in widths))
        for user, name, desc in napps:
            desc = (desc[:desc_w-3] + '...') if len(desc) > desc_w else desc
            print(row.format(user, name, desc))

        print('\nStatus: (i)nstalled, (e)nabled\n')

    @staticmethod
    def delete(args):
        """Delete NApps from server."""
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            log.info('Deleting NApp %s from server...', mgr.napp_id)
            try:
                mgr.delete()
                log.info('  Deleted.')
            except requests.HTTPError as e:
                if e.response.status_code == 405:
                    log.error('Delete Napp is not allowed yet.')
                else:
                    msg = json.loads(e.response.content)
                    log.error('  Server error: %s - ', msg['error'])


1			"""Translate cli commands to non-cli code."""
2			import json
3			import logging
4			import os
5			import re
6			from urllib.error import HTTPError, URLError
7
8			import requests
9
10			from kytos.utils.napps import NAppsManager
11
12			log = logging.getLogger(__name__)
13
14
15			class NAppsAPI:
16			"""An API for the command-line interface.
17
18			Use the config file only for required options. Static methods are called
19			by the parser and they instantiate an object of this class to fulfill the
20			request.
21			"""
22
23			@classmethod
24			def disable(cls, args):
25			"""Disable subcommand."""
26			mgr = NAppsManager()
27
28			if args['all']:
29			napps = mgr.get_enabled()
30			else:
31			napps = args['<napp>']
32
33			for napp in napps:
34			mgr.set_napp(*napp)
35			log.info('NApp %s:', mgr.napp_id)
36			cls.disable_napp(mgr)
37
38			@staticmethod
39			def disable_napp(mgr):
40			"""Disable a NApp."""
41			if mgr.is_enabled():
42			log.info(' Disabling...')
43			mgr.disable()
44			log.info(' Disabled.')
45
46			@classmethod
47			def enable(cls, args):
48			"""Enable subcommand."""
49			mgr = NAppsManager()
50
51			if args['all']:
52			napps = mgr.get_disabled()
53			else:
54			napps = args['<napp>']
55
56			cls.enable_napps(napps)
57
58			@classmethod
59			def enable_napp(cls, mgr):
60			"""Install one NApp using NAppManager object."""
61			try:
62			if not mgr.is_enabled():
63			log.info(' Enabling...')
64			mgr.enable()
65			log.info(' Enabled.')
66			cls.enable_napps(mgr.dependencies())
67			except (FileNotFoundError, PermissionError) as e:
68			log.error(' %s', e)
69
70			@classmethod
71			def enable_napps(cls, napps):
72			"""Enable a list of NApps.
73
74			Args:
75			napps (list): List of NApps.
76			"""
77			mgr = NAppsManager()
78			for napp in napps:
79			mgr.set_napp(*napp)
80			log.info('NApp %s:', mgr.napp_id)
81			cls.enable_napp(mgr)
82
83			@classmethod
84			def create(cls, args):
85			"""Bootstrap a basic NApp structure on the current folder."""
86			NAppsManager.create_napp()
87
88			@classmethod
89			def upload(cls, args):
90			"""Upload the NApp to the NApps server.
91
92			Create the NApp package and upload it to the NApp server.
93			"""
94			try:
95			NAppsManager().upload()
96			except FileNotFoundError:
97			log.error("Couldn't find kytos.json in current directory.")
98
99			@classmethod
100			def uninstall(cls, args):
101			"""Uninstall and delete NApps.
102
103			For local installations, do not delete code outside install_path and
104			enabled_path.
105			"""
106			mgr = NAppsManager()
107			for napp in args['<napp>']:
108			mgr.set_napp(*napp)
109			log.info('NApp %s:', mgr.napp_id)
110			if mgr.is_installed():
111			if mgr.is_enabled():
112			cls.disable_napp(mgr)
113			log.info(' Uninstalling...')
114			mgr.uninstall()
115			log.info(' Uninstalled.')
116
117			@classmethod
118			def install(cls, args):
119			"""Install local or remote NApps."""
120			cls.install_napps(args['<napp>'])
121
122			@classmethod
123			def install_napps(cls, napps):
124			"""Install local or remote NApps.
125
126			This method is recursive, it will install each napps and your
127			dependencies.
128			"""
129			mgr = NAppsManager()
130			for napp in napps:
131			mgr.set_napp(*napp)
132			log.info('NApp %s:', mgr.napp_id)
133			if not mgr.is_installed():
134			cls.install_napp(mgr)
135			else:
136			log.info(' Installed.')
137			cls.enable_napp(mgr)
138			napp_dependencies = mgr.dependencies()
139			if napp_dependencies:
140			log.info('Installing Dependencies:')
141			cls.install_napps(napp_dependencies)
142
143			@classmethod
144			def install_napp(cls, mgr):
145			"""Install a NApp."""
146			try:
147			log.info(' Searching local NApp...')
148			mgr.install_local()
149			log.info(' Found and installed.')
150			cls.enable_napp(mgr)
151			except FileNotFoundError:
152			log.info(' Not found. Downloading from NApps Server...')
153			try:
154			mgr.install_remote()
155			log.info(' Downloaded and installed.')
156			cls.enable_napp(mgr)
157			except HTTPError as e:
158			if e.code == 404:
159			log.error(' NApp not found.')
160			else:
161			log.error(' NApps Server error: %s', e)
162			except URLError as e:
163			log.error(' NApps Server error: '+str(e.reason))
164
165			@classmethod
166			def search(cls, args):
167			"""Search for NApps in NApps server matching a pattern."""
168			safe_shell_pat = re.escape(args['<pattern>']).replace(r'\', '.')
169			pat_str = '.{}.'.format(safe_shell_pat)
170			pattern = re.compile(pat_str, re.IGNORECASE)
171			remote_json = NAppsManager.search(pattern)
172			remote = set()
173			for napp in remote_json:
174			# WARNING: This will be changed in future versions, when 'author'
175			# will be removed.
176			username = napp.get('username', napp.get('author'))
177			remote.add(((username, napp.get('name')), napp.get('description')))
178
179			cls._print_napps(remote)
180
181			@classmethod
182			def _print_napps(cls, napp_list):
183			"""Format the NApp list to be printed."""
184			mgr = NAppsManager()
185			enabled = mgr.get_enabled()
186			installed = mgr.get_installed()
187			napps = []
188			for napp, desc in sorted(napp_list):
189			status = 'i' if napp in installed else '-'
190			status += 'e' if napp in enabled else '-'
191			status = '[{}]'.format(status)
192			name = '{}/{}'.format(*napp)
193			napps.append((status, name, desc))
194			cls.print_napps(napps)
195
196			@classmethod
197			def list(cls, args):
198			"""List all installed NApps and inform whether they are enabled."""
199			mgr = NAppsManager()
200
201			# Add status
202			napps = [napp + ('[ie]',) for napp in mgr.get_enabled()]
203			napps += [napp + ('[i-]',) for napp in mgr.get_disabled()]
204
205			# Sort, add description and reorder coloumns
206			napps.sort()
207			napps_ordered = []
208			for user, name, status in napps:
209			napps_ordered.append((status, '{}/{}'.format(user, name),
210			mgr.get_description(user, name)))
211
212			cls.print_napps(napps_ordered)
213
214			@staticmethod
215			def print_napps(napps):
216			"""Print status, name and description."""
217			if not napps:
218			print('No NApps found.')
219			return
220
221			stat_w = 6 # We already know the size of Status col
222			name_w = max(len(n[1]) for n in napps)
223			desc_w = max(len(n[2]) for n in napps)
224			term_w = os.popen('stty size', 'r').read().split()[1]
225			remaining = int(term_w) - stat_w - name_w - 6
226			desc_w = min(desc_w, remaining)
227			widths = (stat_w, name_w, desc_w)
228
229			header = '\n{:^%d} \| {:^%d} \| {:^%d}' % widths
230			row = '{:^%d} \| {:<%d} \| {:<%d}' % widths
231			print(header.format('Status', 'NApp ID', 'Description'))
232			print('=+='.join('=' * w for w in widths))
233			for user, name, desc in napps:
234			desc = (desc[:desc_w-3] + '...') if len(desc) > desc_w else desc
235			print(row.format(user, name, desc))
236
237			print('\nStatus: (i)nstalled, (e)nabled\n')
238
239			@staticmethod
240			def delete(args):
241			"""Delete NApps from server."""
242			mgr = NAppsManager()
243			for napp in args['<napp>']:
244			mgr.set_napp(*napp)
245			log.info('Deleting NApp %s from server...', mgr.napp_id)
246			try:
247			mgr.delete()
248			log.info(' Deleted.')
249			except requests.HTTPError as e:
250			if e.response.status_code == 405:
251			log.error('Delete Napp is not allowed yet.')
252			else:
253			msg = json.loads(e.response.content)
254			log.error(' Server error: %s - ', msg['error'])
255

kytos / kytos-utils

Pull Request — master (#110)

NAppsAPI.list() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like