ContentPackLoader.get_packs() - Code Metrics - Inspection of "Merge pull request #2824 from StackStorm/pluggable..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( b8d4ca...421369 )

by Plexxi

created 2016-10-13 10:00 UTC

ContentPackLoader.get_packs() A

↳ Parent: ContentPackLoader

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
dl	0
loc	19
rs	9.2
c	0
b	0
f	0

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import re

from yaml.parser import ParserError
import six

from st2common import log as logging
from st2common.constants.meta import ALLOWED_EXTS
from st2common.constants.meta import PARSER_FUNCS
from st2common.constants.pack import MANIFEST_FILE_NAME
from st2common.constants.runners import MANIFEST_FILE_NAME as RUNNER_MANIFEST_FILE_NAME
from st2common.constants.runners import RUNNER_NAME_WHITELIST

__all__ = [
    'RunnersLoader',
    'ContentPackLoader',
    'MetaLoader'
]

LOG = logging.getLogger(__name__)


class RunnersLoader(object):
    """Class for loading runners from directories on disk.
    """

    def get_runners(self, base_dirs):
        """Retrieve a list of runners in the provided directories.

        :return: Dictionary where the key is runner name and the value is full path to the runner
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
            result.update(runners_in_dir)

        return result

    def _get_runners_from_dir(self, base_dir):
        result = {}
        for runner_name in os.listdir(base_dir):
            runner_dir = os.path.join(base_dir, runner_name)
            runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)

            if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
                result[runner_name] = runner_dir

        return result


class ContentPackLoader(object):
    """
    Class for loading pack and pack content information from directories on disk.
    """

    # TODO: Rename "get_content" methods since they don't actually return
    # content - they just return a path

    ALLOWED_CONTENT_TYPES = [
        'triggers',
        'sensors',
        'actions',
        'rules',
        'aliases',
        'policies'
    ]

    def get_packs(self, base_dirs):
        """
        Retrieve a list of packs in the provided directories.

        :return: Dictionary where the key is pack name and the value is full path to the pack
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            packs_in_dir = self._get_packs_from_dir(base_dir=base_dir)
            result.update(packs_in_dir)

        return result

    def get_content(self, base_dirs, content_type):
        """
        Retrieve content from the provided directories.

        Provided directories are searched from left to right. If a pack with the same name exists
        in multiple directories, first pack which is found wins.

        :param base_dirs: Directories to look into.
        :type base_dirs: ``list``

        :param content_type: Content type to look for (sensors, actions, rules).
        :type content_type: ``str``

        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        if content_type not in self.ALLOWED_CONTENT_TYPES:
            raise ValueError('Unsupported content_type: %s' % (content_type))

        content = {}
        pack_to_dir_map = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            dir_content = self._get_content_from_dir(base_dir=base_dir, content_type=content_type)

            # Check for duplicate packs
            for pack_name, pack_content in six.iteritems(dir_content):
                if pack_name in content:
                    pack_dir = pack_to_dir_map[pack_name]
                    LOG.warning('Pack "%s" already found in "%s", ignoring content from "%s"' %

                                (pack_name, pack_dir, base_dir))
                else:
                    content[pack_name] = pack_content
                    pack_to_dir_map[pack_name] = base_dir

        return content

    def get_content_from_pack(self, pack_dir, content_type):
        """
        Retrieve content from the provided pack directory.

        :param pack_dir: Path to the pack directory.
        :type pack_dir: ``str``

        :param content_type: Content type to look for (sensors, actions, rules).
        :type content_type: ``str``

        :rtype: ``str``
        """
        if content_type not in self.ALLOWED_CONTENT_TYPES:
            raise ValueError('Unsupported content_type: %s' % (content_type))

        if not os.path.isdir(pack_dir):
            raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))

        content = self._get_content_from_pack_dir(pack_dir=pack_dir,
                                                  content_type=content_type)
        return content

    def _get_packs_from_dir(self, base_dir):
        result = {}
        for pack_name in os.listdir(base_dir):
            pack_dir = os.path.join(base_dir, pack_name)
            pack_manifest_file = os.path.join(pack_dir, MANIFEST_FILE_NAME)

            if os.path.isdir(pack_dir) and os.path.isfile(pack_manifest_file):
                result[pack_name] = pack_dir

        return result

    def _get_content_from_dir(self, base_dir, content_type):
        content = {}
        for pack in os.listdir(base_dir):
            # TODO: Use function from util which escapes the name
            pack_dir = os.path.join(base_dir, pack)

            # Ignore missing or non directories
            try:
                pack_content = self._get_content_from_pack_dir(pack_dir=pack_dir,
                                                               content_type=content_type)
            except ValueError:
                continue
            else:
                content[pack] = pack_content

        return content

    def _get_content_from_pack_dir(self, pack_dir, content_type):
        content_types = dict(
            triggers=self._get_triggers,
            sensors=self._get_sensors,
            actions=self._get_actions,
            rules=self._get_rules,
            aliases=self._get_aliases,
            policies=self._get_policies
        )

        get_func = content_types.get(content_type)

        if get_func is None:
            raise ValueError('Invalid content_type: %s' % (content_type))

        if not os.path.isdir(pack_dir):
            raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))

        pack_content = get_func(pack_dir=pack_dir)
        return pack_content

    def _get_triggers(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='triggers')

    def _get_sensors(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='sensors')

    def _get_actions(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='actions')

    def _get_rules(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='rules')

    def _get_aliases(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='aliases')

    def _get_policies(self, pack_dir):
        return self._get_folder(pack_dir=pack_dir, content_type='policies')

    def _get_folder(self, pack_dir, content_type):
        path = os.path.join(pack_dir, content_type)
        if not os.path.isdir(path):
            return None
        return path

    def _get_runners_from_dir(self, base_dir):
        result = {}
        for runner_name in os.listdir(base_dir):
            if not re.match(RUNNER_NAME_WHITELIST, runner_name):
                raise ValueError('Invalid runner name "%s"' % (runner_name))

            runner_dir = os.path.join(base_dir, runner_name)
            runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)

            if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
                LOG.debug("Loading runner manifest for: %s" % (runner_name))

                result[runner_name] = runner_dir
            else:
                LOG.debug("Could not load manifest for runner: %s" % (runner_name))


        return result

    def get_runners(self, base_dirs):
        """Retrieve a list of runners in the provided directories.

        :return: Dictionary where the key is runner name and the value is full path to the runner
                 directory.
        :rtype: ``dict``
        """
        assert isinstance(base_dirs, list)

        result = {}
        for base_dir in base_dirs:
            if not os.path.isdir(base_dir):
                raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))

            runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
            result.update(runners_in_dir)

        return result


class MetaLoader(object):
    """
    Class for loading and parsing pack and resource metadata files.
    """

    def load(self, file_path, expected_type=None):
        """
        Loads content from file_path if file_path's extension
        is one of allowed ones (See ALLOWED_EXTS).

        Throws UnsupportedMetaException on disallowed filetypes.
        Throws ValueError on malformed meta.

        :param file_path: Absolute path to the file to load content from.
        :type file_path: ``str``

        :param expected_type: Expected type for the loaded and parsed content (optional).
        :type expected_type: ``object``

        :rtype: ``dict``
        """
        file_name, file_ext = os.path.splitext(file_path)

        if file_ext not in ALLOWED_EXTS:
            raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
                            (file_ext, file_path, ALLOWED_EXTS))

        result = self._load(PARSER_FUNCS[file_ext], file_path)

        if expected_type and not isinstance(result, expected_type):
            actual_type = type(result).__name__
            error = 'Expected "%s", got "%s"' % (expected_type.__name__, actual_type)
            raise ValueError(error)

        return result

    def _load(self, parser_func, file_path):
        with open(file_path, 'r') as fd:
            try:
                return parser_func(fd)
            except ValueError:
                LOG.exception('Failed loading content from %s.', file_path)
                raise
            except ParserError:
                LOG.exception('Failed loading content from %s.', file_path)
                raise


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
17			import re
18
19			from yaml.parser import ParserError
20			import six
21
22			from st2common import log as logging
23			from st2common.constants.meta import ALLOWED_EXTS
24			from st2common.constants.meta import PARSER_FUNCS
25			from st2common.constants.pack import MANIFEST_FILE_NAME
26			from st2common.constants.runners import MANIFEST_FILE_NAME as RUNNER_MANIFEST_FILE_NAME
27			from st2common.constants.runners import RUNNER_NAME_WHITELIST
28
29			__all__ = [
30			'RunnersLoader',
31			'ContentPackLoader',
32			'MetaLoader'
33			]
34
35			LOG = logging.getLogger(__name__)
36
37
38			class RunnersLoader(object):
39			"""Class for loading runners from directories on disk.
40			"""
41
42			def get_runners(self, base_dirs):
43			"""Retrieve a list of runners in the provided directories.
44
45			:return: Dictionary where the key is runner name and the value is full path to the runner
46			directory.
47			:rtype: ``dict``
48			"""
49			assert isinstance(base_dirs, list)
50
51			result = {}
52			for base_dir in base_dirs:
53			if not os.path.isdir(base_dir):
54			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
55
56			runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
57			result.update(runners_in_dir)
58
59			return result
60
61			def _get_runners_from_dir(self, base_dir):
62			result = {}
63			for runner_name in os.listdir(base_dir):
64			runner_dir = os.path.join(base_dir, runner_name)
65			runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)
66
67			if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
68			result[runner_name] = runner_dir
69
70			return result
71
72
73			class ContentPackLoader(object):
74			"""
75			Class for loading pack and pack content information from directories on disk.
76			"""
77
78			# TODO: Rename "get_content" methods since they don't actually return
79			# content - they just return a path
80
81			ALLOWED_CONTENT_TYPES = [
82			'triggers',
83			'sensors',
84			'actions',
85			'rules',
86			'aliases',
87			'policies'
88			]
89
90			def get_packs(self, base_dirs):
91			"""
92			Retrieve a list of packs in the provided directories.
93
94			:return: Dictionary where the key is pack name and the value is full path to the pack
95			directory.
96			:rtype: ``dict``
97			"""
98			assert isinstance(base_dirs, list)
99
100			result = {}
101			for base_dir in base_dirs:
102			if not os.path.isdir(base_dir):
103			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
104
105			packs_in_dir = self._get_packs_from_dir(base_dir=base_dir)
106			result.update(packs_in_dir)
107
108			return result
109
110			def get_content(self, base_dirs, content_type):
111			"""
112			Retrieve content from the provided directories.
113
114			Provided directories are searched from left to right. If a pack with the same name exists
115			in multiple directories, first pack which is found wins.
116
117			:param base_dirs: Directories to look into.
118			:type base_dirs: ``list``
119
120			:param content_type: Content type to look for (sensors, actions, rules).
121			:type content_type: ``str``
122
123			:rtype: ``dict``
124			"""
125			assert isinstance(base_dirs, list)
126
127			if content_type not in self.ALLOWED_CONTENT_TYPES:
128			raise ValueError('Unsupported content_type: %s' % (content_type))
129
130			content = {}
131			pack_to_dir_map = {}
132			for base_dir in base_dirs:
133			if not os.path.isdir(base_dir):
134			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
135
136			dir_content = self._get_content_from_dir(base_dir=base_dir, content_type=content_type)
137
138			# Check for duplicate packs
139			for pack_name, pack_content in six.iteritems(dir_content):
140			if pack_name in content:
141			pack_dir = pack_to_dir_map[pack_name]
142			LOG.warning('Pack "%s" already found in "%s", ignoring content from "%s"' %
			0 ignored issues – show Coding Style Best Practice introduced 2015-11-20 23:04 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
143			(pack_name, pack_dir, base_dir))
144			else:
145			content[pack_name] = pack_content
146			pack_to_dir_map[pack_name] = base_dir
147
148			return content
149
150			def get_content_from_pack(self, pack_dir, content_type):
151			"""
152			Retrieve content from the provided pack directory.
153
154			:param pack_dir: Path to the pack directory.
155			:type pack_dir: ``str``
156
157			:param content_type: Content type to look for (sensors, actions, rules).
158			:type content_type: ``str``
159
160			:rtype: ``str``
161			"""
162			if content_type not in self.ALLOWED_CONTENT_TYPES:
163			raise ValueError('Unsupported content_type: %s' % (content_type))
164
165			if not os.path.isdir(pack_dir):
166			raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))
167
168			content = self._get_content_from_pack_dir(pack_dir=pack_dir,
169			content_type=content_type)
170			return content
171
172			def _get_packs_from_dir(self, base_dir):
173			result = {}
174			for pack_name in os.listdir(base_dir):
175			pack_dir = os.path.join(base_dir, pack_name)
176			pack_manifest_file = os.path.join(pack_dir, MANIFEST_FILE_NAME)
177
178			if os.path.isdir(pack_dir) and os.path.isfile(pack_manifest_file):
179			result[pack_name] = pack_dir
180
181			return result
182
183			def _get_content_from_dir(self, base_dir, content_type):
184			content = {}
185			for pack in os.listdir(base_dir):
186			# TODO: Use function from util which escapes the name
187			pack_dir = os.path.join(base_dir, pack)
188
189			# Ignore missing or non directories
190			try:
191			pack_content = self._get_content_from_pack_dir(pack_dir=pack_dir,
192			content_type=content_type)
193			except ValueError:
194			continue
195			else:
196			content[pack] = pack_content
197
198			return content
199
200			def _get_content_from_pack_dir(self, pack_dir, content_type):
201			content_types = dict(
202			triggers=self._get_triggers,
203			sensors=self._get_sensors,
204			actions=self._get_actions,
205			rules=self._get_rules,
206			aliases=self._get_aliases,
207			policies=self._get_policies
208			)
209
210			get_func = content_types.get(content_type)
211
212			if get_func is None:
213			raise ValueError('Invalid content_type: %s' % (content_type))
214
215			if not os.path.isdir(pack_dir):
216			raise ValueError('Directory "%s" doesn\'t exist' % (pack_dir))
217
218			pack_content = get_func(pack_dir=pack_dir)
219			return pack_content
220
221			def _get_triggers(self, pack_dir):
222			return self._get_folder(pack_dir=pack_dir, content_type='triggers')
223
224			def _get_sensors(self, pack_dir):
225			return self._get_folder(pack_dir=pack_dir, content_type='sensors')
226
227			def _get_actions(self, pack_dir):
228			return self._get_folder(pack_dir=pack_dir, content_type='actions')
229
230			def _get_rules(self, pack_dir):
231			return self._get_folder(pack_dir=pack_dir, content_type='rules')
232
233			def _get_aliases(self, pack_dir):
234			return self._get_folder(pack_dir=pack_dir, content_type='aliases')
235
236			def _get_policies(self, pack_dir):
237			return self._get_folder(pack_dir=pack_dir, content_type='policies')
238
239			def _get_folder(self, pack_dir, content_type):
240			path = os.path.join(pack_dir, content_type)
241			if not os.path.isdir(path):
242			return None
243			return path
244
245			def _get_runners_from_dir(self, base_dir):
246			result = {}
247			for runner_name in os.listdir(base_dir):
248			if not re.match(RUNNER_NAME_WHITELIST, runner_name):
249			raise ValueError('Invalid runner name "%s"' % (runner_name))
250
251			runner_dir = os.path.join(base_dir, runner_name)
252			runner_manifest_file = os.path.join(runner_dir, RUNNER_MANIFEST_FILE_NAME)
253
254			if os.path.isdir(runner_dir) and os.path.isfile(runner_manifest_file):
255			LOG.debug("Loading runner manifest for: %s" % (runner_name))
			0 ignored issues – show Coding Style Best Practice introduced 2016-09-28 00:35 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
256			result[runner_name] = runner_dir
257			else:
258			LOG.debug("Could not load manifest for runner: %s" % (runner_name))
			0 ignored issues – show Coding Style Best Practice introduced 2016-09-28 00:35 UTC by Report Bug Copy Issue Report Specify string format arguments as logging function parameters Loading history...
259
260			return result
261
262			def get_runners(self, base_dirs):
263			"""Retrieve a list of runners in the provided directories.
264
265			:return: Dictionary where the key is runner name and the value is full path to the runner
266			directory.
267			:rtype: ``dict``
268			"""
269			assert isinstance(base_dirs, list)
270
271			result = {}
272			for base_dir in base_dirs:
273			if not os.path.isdir(base_dir):
274			raise ValueError('Directory "%s" doesn\'t exist' % (base_dir))
275
276			runners_in_dir = self._get_runners_from_dir(base_dir=base_dir)
277			result.update(runners_in_dir)
278
279			return result
280
281
282			class MetaLoader(object):
283			"""
284			Class for loading and parsing pack and resource metadata files.
285			"""
286
287			def load(self, file_path, expected_type=None):
288			"""
289			Loads content from file_path if file_path's extension
290			is one of allowed ones (See ALLOWED_EXTS).
291
292			Throws UnsupportedMetaException on disallowed filetypes.
293			Throws ValueError on malformed meta.
294
295			:param file_path: Absolute path to the file to load content from.
296			:type file_path: ``str``
297
298			:param expected_type: Expected type for the loaded and parsed content (optional).
299			:type expected_type: ``object``
300
301			:rtype: ``dict``
302			"""
303			file_name, file_ext = os.path.splitext(file_path)
304
305			if file_ext not in ALLOWED_EXTS:
306			raise Exception('Unsupported meta type %s, file %s. Allowed: %s' %
307			(file_ext, file_path, ALLOWED_EXTS))
308
309			result = self._load(PARSER_FUNCS[file_ext], file_path)
310
311			if expected_type and not isinstance(result, expected_type):
312			actual_type = type(result).__name__
313			error = 'Expected "%s", got "%s"' % (expected_type.__name__, actual_type)
314			raise ValueError(error)
315
316			return result
317
318			def _load(self, parser_func, file_path):
319			with open(file_path, 'r') as fd:
320			try:
321			return parser_func(fd)
322			except ValueError:
323			LOG.exception('Failed loading content from %s.', file_path)
324			raise
325			except ParserError:
326			LOG.exception('Failed loading content from %s.', file_path)
327			raise
328

StackStorm / st2

Push — develop ( b8d4ca...421369 )

ContentPackLoader.get_packs() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like