generate_pack() - Code Metrics - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

generate_pack() F
last analyzed 2017-07-26 18:29 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
dl	0
loc	92
rs	3.1764
c	0
b	0
f	0

How to fix Long Method Complexity

"""Generate pack based on inspecting PyNOS package
"""
import inspect
import re
import xml.etree.ElementTree as et
import pkgutil
import yaml
from docutils import core
import pynos.versions.base


def load_modules():
    """Load modules f or parsing
    """
    for _, modname, _ in pkgutil.iter_modules(pynos.versions.base.__path__):
        __import__('pynos.versions.base.' + modname)


def load_methods():
    """Get methods from modules
    """
    classes = []
    for _, module in inspect.getmembers(
            pynos.versions.base,
            predicate=inspect.ismodule
    ):
        classes.append(inspect.getmembers(module, predicate=inspect.isclass))
    methods = {}
    for cls in classes:
        if isinstance(cls, list) and len(cls) > 0:
            cls = cls[0]
        else:
            continue
        methods.update(
            {
                cls[0]: inspect.getmembers(cls[1], predicate=inspect.ismethod)
            }
        )

    return methods


def parse_methods(methods):
    """Parse methods extract docstring and parse dockstring to get parameters
    """
    info = {}
    for name, methods in methods.iteritems():
        method_args = dict()
        for method in methods:
            if method[0][0] == "_":
                continue
            try:
                xml_docstring = et.fromstring(
                    core.publish_doctree(
                        method[1].im_func.func_doc
                    ).asdom().toxml()
                )
                arguments_block = xml_docstring.find('block_quote').\
                    find('definition_list').find('definition_list_item').\
                    find('definition').find('paragraph').text
            except AttributeError, error:
                print error
                continue
            seperated_arguments = arguments_block.split('\n')
            arg = []
            for item in seperated_arguments:
                items = re.findall(r'\s*([A-z0-9_]*)\s*\(([A-z0-9]*)\):'
                                   r'\s*(.*)', item)
                if len(items) > 0:
                    if items[0][0] == "callback":
                        continue
                    arg.append(items[0])
            if xml_docstring.find('paragraph') is not None:
                arg.append(
                    (
                        '__description__',
                        xml_docstring.find('paragraph').text
                    )
                )
            method_args.update({method[0]: arg})
        info.update({name: method_args})
    return info


def generate_pack(info):
    """Generate pack and write to disk.
    """
    for module, methods in info.iteritems():
        for method, args in methods.iteritems():
            action_name = "%s_%s" % (str.lower(module), method)
            action_path = "%s.%s" % (str.lower(module), method)
            code = """from pynos import device
from st2actions.runners.pythonrunner import Action


class %s(Action):
    def run(self, **kwargs):
        conn = (str(kwargs.pop('ip')), str(kwargs.pop('port')))
        auth = (str(kwargs.pop('username')), str(kwargs.pop('password')))
        test = kwargs.pop('test', False)
        callback = kwargs.pop('callback', None)
        with device.Device(
            conn=conn, auth=auth,
            test=test,
            callback=callback
        ) as dev:
            dev.%s(**kwargs)
        return 0\n""" % (action_name, action_path)
            action_yaml = {
                'name': action_name,
                'runner_type': "python-script",
                'description': "",
                'enabled': True,
                'entry_point': "%s.py" % action_name,
                'parameters': {
                    'ip': {
                        'type': 'string',
                        'description': 'IP address of VDX to connect to.',
                        'required': True,
                        'position': 0
                    },
                    'port': {
                        'type': 'string',
                        'description': 'Port to use to connect to VDX.',
                        'required': True,
                        'default': '22',
                        'position': 1
                    },
                    'username': {
                        'type': 'string',
                        'description': 'Username used with authentication.',
                        'required': True,
                        'position': 2
                    },
                    'password': {
                        'type': 'string',
                        'description': 'Password used with authentication.',
                        'required': True,
                        'secret': True,
                        'position': 3
                    }
                }
            }
            type_map = {
                'str': 'string',
                'bool': 'boolean',
                'int': 'integer',
                'tuple(str, str)': 'array'
            }
            if len(args) > 0:
                position = 4
                for arg in args:
                    if arg[0] == '__description__':
                        action_yaml['description'] = arg[1]
                        continue
                    if arg[1] == 'function':
                        continue
                    if type_map.get(arg[1]) is None:
                        raise Exception("%s maps to none" % arg[1])
                    yaml_arg = {
                        'type': type_map.get(arg[1]),
                        'description': arg[2],
                        'required': True,
                        'position': position
                    }
                    if arg[1] == 'bool':
                        del yaml_arg['required']
                    position += 1
                    action_yaml['parameters'].update({arg[0]: yaml_arg})

            with open("pack/actions/%s.py" % action_name, "w+") as py_file:
                py_file.write(code)

            with open("pack/actions/%s.yaml" % action_name, "w+") as py_file:
                output = yaml.dump(action_yaml, default_flow_style=False)
                py_file.write(output)


def main():
    """Main entry point
    """
    load_modules()
    methods = load_methods()
    info = parse_methods(methods)
    generate_pack(info)


if __name__ == "__main__":
    main()


1			"""Generate pack based on inspecting PyNOS package
2			"""
3			import inspect
4			import re
5			import xml.etree.ElementTree as et
6			import pkgutil
7			import yaml
8			from docutils import core
9			import pynos.versions.base
10
11
12			def load_modules():
13			"""Load modules f or parsing
14			"""
15			for _, modname, _ in pkgutil.iter_modules(pynos.versions.base.__path__):
16			__import__('pynos.versions.base.' + modname)
17
18
19			def load_methods():
20			"""Get methods from modules
21			"""
22			classes = []
23			for _, module in inspect.getmembers(
24			pynos.versions.base,
25			predicate=inspect.ismodule
26			):
27			classes.append(inspect.getmembers(module, predicate=inspect.isclass))
28			methods = {}
29			for cls in classes:
30			if isinstance(cls, list) and len(cls) > 0:
31			cls = cls[0]
32			else:
33			continue
34			methods.update(
35			{
36			cls[0]: inspect.getmembers(cls[1], predicate=inspect.ismethod)
37			}
38			)
39
40			return methods
41
42
43			def parse_methods(methods):
44			"""Parse methods extract docstring and parse dockstring to get parameters
45			"""
46			info = {}
47			for name, methods in methods.iteritems():
48			method_args = dict()
49			for method in methods:
50			if method[0][0] == "_":
51			continue
52			try:
53			xml_docstring = et.fromstring(
54			core.publish_doctree(
55			method[1].im_func.func_doc
56			).asdom().toxml()
57			)
58			arguments_block = xml_docstring.find('block_quote').\
59			find('definition_list').find('definition_list_item').\
60			find('definition').find('paragraph').text
61			except AttributeError, error:
62			print error
63			continue
64			seperated_arguments = arguments_block.split('\n')
65			arg = []
66			for item in seperated_arguments:
67			items = re.findall(r'\s([A-z0-9_])\s\(([A-z0-9])\):'
68			r'\s(.)', item)
69			if len(items) > 0:
70			if items[0][0] == "callback":
71			continue
72			arg.append(items[0])
73			if xml_docstring.find('paragraph') is not None:
74			arg.append(
75			(
76			'__description__',
77			xml_docstring.find('paragraph').text
78			)
79			)
80			method_args.update({method[0]: arg})
81			info.update({name: method_args})
82			return info
83
84
85			def generate_pack(info):
86			"""Generate pack and write to disk.
87			"""
88			for module, methods in info.iteritems():
89			for method, args in methods.iteritems():
90			action_name = "%s_%s" % (str.lower(module), method)
91			action_path = "%s.%s" % (str.lower(module), method)
92			code = """from pynos import device
93			from st2actions.runners.pythonrunner import Action
94
95
96			class %s(Action):
97			def run(self, **kwargs):
98			conn = (str(kwargs.pop('ip')), str(kwargs.pop('port')))
99			auth = (str(kwargs.pop('username')), str(kwargs.pop('password')))
100			test = kwargs.pop('test', False)
101			callback = kwargs.pop('callback', None)
102			with device.Device(
103			conn=conn, auth=auth,
104			test=test,
105			callback=callback
106			) as dev:
107			dev.%s(**kwargs)
108			return 0\n""" % (action_name, action_path)
109			action_yaml = {
110			'name': action_name,
111			'runner_type': "python-script",
112			'description': "",
113			'enabled': True,
114			'entry_point': "%s.py" % action_name,
115			'parameters': {
116			'ip': {
117			'type': 'string',
118			'description': 'IP address of VDX to connect to.',
119			'required': True,
120			'position': 0
121			},
122			'port': {
123			'type': 'string',
124			'description': 'Port to use to connect to VDX.',
125			'required': True,
126			'default': '22',
127			'position': 1
128			},
129			'username': {
130			'type': 'string',
131			'description': 'Username used with authentication.',
132			'required': True,
133			'position': 2
134			},
135			'password': {
136			'type': 'string',
137			'description': 'Password used with authentication.',
138			'required': True,
139			'secret': True,
140			'position': 3
141			}
142			}
143			}
144			type_map = {
145			'str': 'string',
146			'bool': 'boolean',
147			'int': 'integer',
148			'tuple(str, str)': 'array'
149			}
150			if len(args) > 0:
151			position = 4
152			for arg in args:
153			if arg[0] == '__description__':
154			action_yaml['description'] = arg[1]
155			continue
156			if arg[1] == 'function':
157			continue
158			if type_map.get(arg[1]) is None:
159			raise Exception("%s maps to none" % arg[1])
160			yaml_arg = {
161			'type': type_map.get(arg[1]),
162			'description': arg[2],
163			'required': True,
164			'position': position
165			}
166			if arg[1] == 'bool':
167			del yaml_arg['required']
168			position += 1
169			action_yaml['parameters'].update({arg[0]: yaml_arg})
170
171			with open("pack/actions/%s.py" % action_name, "w+") as py_file:
172			py_file.write(code)
173
174			with open("pack/actions/%s.yaml" % action_name, "w+") as py_file:
175			output = yaml.dump(action_yaml, default_flow_style=False)
176			py_file.write(output)
177
178
179			def main():
180			"""Main entry point
181			"""
182			load_modules()
183			methods = load_methods()
184			info = parse_methods(methods)
185			generate_pack(info)
186
187
188			if __name__ == "__main__":
189			main()
190

StackStorm / st2contrib

generate_pack() F last analyzed 2017-07-26 18:29 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

generate_pack() F
last analyzed 2017-07-26 18:29 UTC