ActivityStep.__init__() - Code Metrics - Inspection of "Added WORKFLOW_COMPLETED message for SQS" - sportarchive/CloudProcessingEngine-Decider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#35)

by Tyler

created 2016-05-20 20:56 UTC

ActivityStep.init() B

↳ Parent: ActivityStep

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	5
dl	0
loc	21
rs	8.439
c	2
b	0
f	0

from __future__ import (
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    absolute_import,
    division,
    print_function
)

import abc
import json
import logging

import jinja2
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
import jinja2.meta
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from .state_status import StepStateStatus
from .step_results import (
    ActivityStepResult,
    TemplatedStepResult,
)

_LOGGER = logging.getLogger(__name__)


class StepError(StandardError):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    pass


class StepDefinitionError(StepError):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    pass


class Step(object):
    """Base `Step` in a workflow.

    A `Step` is composed of a name and a collection of its requirements.
    """

    __slots__ = ('name', 'requires')
    __metaclass__ = abc.ABCMeta

    def __init__(self, name, requires=()):
        self.name = name
        self.requires = dict([self._resolve_parent(parent_def)
                              for parent_def in requires])

    @staticmethod
    def _resolve_parent(parent_def):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""

        if isinstance(parent_def, basestring):

            return (parent_def, StepStateStatus.completed)
        else:
            try:
                parent_name, status_name = parent_def
                return (parent_name, getattr(StepStateStatus, status_name))

            except AttributeError as err:
                raise StepDefinitionError('Invalid Step status name: %r' %

                                          err.args[0])
            except StandardError:

                raise StepDefinitionError('Invalid Step definition: %r' %

                                          parent_def)

    @classmethod
    def from_data(cls, step_data, activities):
        """Create a new Step object from a step definition"""
        # FIXME: pass data through JSONSchema

        if 'activity' in step_data:
            activity_name = step_data['activity']
            activity = activities[activity_name]

            step = ActivityStep(
                name=step_data['name'],
                requires=step_data.get('requires', ()),
                activity=activity,
                input_template=step_data.get('input', None),
            )

        elif 'eval' in step_data:
            step = TemplatedStep(
                name=step_data['name'],
                requires=step_data.get('requires', ()),
                eval_block=step_data['eval'],
            )

        else:
            raise ValueError('Invalid Step data: %r' % step_data)

        return step

    @abc.abstractmethod
    def prepare(self, _context):
        """Prepare step's input from context.

        :returns:
            Step's input
        """
        pass

    @abc.abstractmethod
    def run(self, _step_input):
        """Run the step from its input.

        :returns:
            A StepResult object
        """
        pass

    @abc.abstractmethod
    def render(self, _output):
        """Renders the Step's output for usage by other steps.

        :returns:
            A dictionary of 'key: value' pairs.
        """
        pass

    def __repr__(self):
        return '{ctype}(name={name})'.format(ctype=self.__class__.__name__,
                                             name=self.name)


class ActivityStep(Step):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""

    __slots__ = ('activity', 'input_template', '__env')

    def __init__(self, name, activity, input_template, requires=()):
        super(ActivityStep, self).__init__(name, requires)
        self.activity = activity
        self.input_template = None

        # Define the Jinga2 environment
        self.__env = jinja2.Environment(finalize=self.__jsonify)

        if input_template is not None:
            tp_required = self._check_template_dependencies(input_template)
            for tp_var in tp_required:
                # `__input__` is a "magic" step referencing the workflow input
                if tp_var == '__input__':
                    continue
                if tp_var not in self.requires:
                    raise StepDefinitionError(

                        'Invalid step %r: Template used %r is not required' %
                        (self.name, tp_var,)
                    )

            self.input_template = self.__env.from_string(input_template)

    def __jsonify(self, obj):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        # Note: This method is called by Jinja2 to "finalize" its variables.
        #       Jinja2 also calls it to "finalize" the non-variable chunks
        #       of a template's string when compiling the template, so we have
        #       to be careful of extra quoting.
        if self.input_template is not None:
            # Now that the templates are defined, we can define the "finalize"
            # function to JSON dump all template variables.
            return json.dumps(obj, indent=4, sort_keys=True)
        else:
            return obj

    def prepare(self, context):
        if self.input_template is not None:
            activity_input_json = self.input_template.render(context)
            # FIXME: We are assuming JSON activity input here

            try:
                activity_input = json.loads(activity_input_json)
            except ValueError:
                _LOGGER.exception('Invalid template result: %r',
                                  activity_input_json)
                raise
        else:
            activity_input = None

        self.activity.check_input(activity_input)
        return activity_input

    def run(self, step_input):
        return ActivityStepResult(
            name=self.name,
            activity=self.activity,
            activity_input=step_input,
        )

    def render(self, output):
        return self.activity.render_outputs(output)

    def _check_template_dependencies(self, input_template):
        """Return the list of used external variable in the template.
        """
        parsed_template = self.__env.parse(input_template)
        return jinja2.meta.find_undeclared_variables(parsed_template)


class TemplatedStep(Step):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""

    __slots__ = ('eval_block')

    def __init__(self, name, eval_block, requires=()):
        super(TemplatedStep, self).__init__(name, requires)
        self.eval_block = jinja2.Template(eval_block)

    def prepare(self, context):
        return self.eval_block.render(context)

    def run(self, _step_input):
        return TemplatedStepResult()

    def render(self, _output):
        # NOTE: Templated steps do not have any usable attributes
        return {}


Pull Request — master (#35)

ActivityStep.init() B

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing init.py files

1. Missing Dependencies

2. Missing init.py files

1			from __future__ import (
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This module should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
2			absolute_import,
3			division,
4			print_function
5			)
6
7			import abc
8			import json
9			import logging
10
11			import jinja2
			0 ignored issues – show Configuration introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report The import `jinja2` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
12			import jinja2.meta
			0 ignored issues – show Configuration introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report The import `jinja2.meta` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
13
14			from .state_status import StepStateStatus
15			from .step_results import (
16			ActivityStepResult,
17			TemplatedStepResult,
18			)
19
20			_LOGGER = logging.getLogger(__name__)
21
22
23			class StepError(StandardError):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history... Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class has no `__init__` method. Loading history... Comprehensibility Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Undefined variable 'StandardError' Loading history...
24			pass
25
26
27			class StepDefinitionError(StepError):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history... Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class has no `__init__` method. Loading history...
28			pass
29
30
31			class Step(object):
32			"""Base `Step` in a workflow.
33
34			A `Step` is composed of a name and a collection of its requirements.
35			"""
36
37			__slots__ = ('name', 'requires')
38			__metaclass__ = abc.ABCMeta
39
40			def __init__(self, name, requires=()):
41			self.name = name
42			self.requires = dict([self._resolve_parent(parent_def)
43			for parent_def in requires])
44
45			@staticmethod
46			def _resolve_parent(parent_def):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
47
48			if isinstance(parent_def, basestring):
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Undefined variable 'basestring' Loading history...
49			return (parent_def, StepStateStatus.completed)
50			else:
51			try:
52			parent_name, status_name = parent_def
53			return (parent_name, getattr(StepStateStatus, status_name))
54
55			except AttributeError as err:
56			raise StepDefinitionError('Invalid Step status name: %r' %
			0 ignored issues – show Bug Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Raising a new style class which doesn't inherit from BaseException Loading history...
57			err.args[0])
58			except StandardError:
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Undefined variable 'StandardError' Loading history...
59			raise StepDefinitionError('Invalid Step definition: %r' %
			0 ignored issues – show Bug Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Raising a new style class which doesn't inherit from BaseException Loading history...
60			parent_def)
61
62			@classmethod
63			def from_data(cls, step_data, activities):
64			"""Create a new Step object from a step definition"""
65			# FIXME: pass data through JSONSchema
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
66			if 'activity' in step_data:
67			activity_name = step_data['activity']
68			activity = activities[activity_name]
69
70			step = ActivityStep(
71			name=step_data['name'],
72			requires=step_data.get('requires', ()),
73			activity=activity,
74			input_template=step_data.get('input', None),
75			)
76
77			elif 'eval' in step_data:
78			step = TemplatedStep(
79			name=step_data['name'],
80			requires=step_data.get('requires', ()),
81			eval_block=step_data['eval'],
82			)
83
84			else:
85			raise ValueError('Invalid Step data: %r' % step_data)
86
87			return step
88
89			@abc.abstractmethod
90			def prepare(self, _context):
91			"""Prepare step's input from context.
92
93			:returns:
94			Step's input
95			"""
96			pass
97
98			@abc.abstractmethod
99			def run(self, _step_input):
100			"""Run the step from its input.
101
102			:returns:
103			A StepResult object
104			"""
105			pass
106
107			@abc.abstractmethod
108			def render(self, _output):
109			"""Renders the Step's output for usage by other steps.
110
111			:returns:
112			A dictionary of 'key: value' pairs.
113			"""
114			pass
115
116			def __repr__(self):
117			return '{ctype}(name={name})'.format(ctype=self.__class__.__name__,
118			name=self.name)
119
120
121			class ActivityStep(Step):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
122
123			__slots__ = ('activity', 'input_template', '__env')
124
125			def __init__(self, name, activity, input_template, requires=()):
126			super(ActivityStep, self).__init__(name, requires)
127			self.activity = activity
128			self.input_template = None
129
130			# Define the Jinga2 environment
131			self.__env = jinja2.Environment(finalize=self.__jsonify)
132
133			if input_template is not None:
134			tp_required = self._check_template_dependencies(input_template)
135			for tp_var in tp_required:
136			# `__input__` is a "magic" step referencing the workflow input
137			if tp_var == '__input__':
138			continue
139			if tp_var not in self.requires:
140			raise StepDefinitionError(
			0 ignored issues – show Bug Best Practice introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Raising a new style class which doesn't inherit from BaseException Loading history...
141			'Invalid step %r: Template used %r is not required' %
142			(self.name, tp_var,)
143			)
144
145			self.input_template = self.__env.from_string(input_template)
146
147			def __jsonify(self, obj):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
148			# Note: This method is called by Jinja2 to "finalize" its variables.
149			# Jinja2 also calls it to "finalize" the non-variable chunks
150			# of a template's string when compiling the template, so we have
151			# to be careful of extra quoting.
152			if self.input_template is not None:
153			# Now that the templates are defined, we can define the "finalize"
154			# function to JSON dump all template variables.
155			return json.dumps(obj, indent=4, sort_keys=True)
156			else:
157			return obj
158
159			def prepare(self, context):
160			if self.input_template is not None:
161			activity_input_json = self.input_template.render(context)
162			# FIXME: We are assuming JSON activity input here
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
163			try:
164			activity_input = json.loads(activity_input_json)
165			except ValueError:
166			_LOGGER.exception('Invalid template result: %r',
167			activity_input_json)
168			raise
169			else:
170			activity_input = None
171
172			self.activity.check_input(activity_input)
173			return activity_input
174
175			def run(self, step_input):
176			return ActivityStepResult(
177			name=self.name,
178			activity=self.activity,
179			activity_input=step_input,
180			)
181
182			def render(self, output):
183			return self.activity.render_outputs(output)
184
185			def _check_template_dependencies(self, input_template):
186			"""Return the list of used external variable in the template.
187			"""
188			parsed_template = self.__env.parse(input_template)
189			return jinja2.meta.find_undeclared_variables(parsed_template)
190
191
192			class TemplatedStep(Step):
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
193
194			__slots__ = ('eval_block')
195
196			def __init__(self, name, eval_block, requires=()):
197			super(TemplatedStep, self).__init__(name, requires)
198			self.eval_block = jinja2.Template(eval_block)
199
200			def prepare(self, context):
201			return self.eval_block.render(context)
202
203			def run(self, _step_input):
204			return TemplatedStepResult()
205
206			def render(self, _output):
207			# NOTE: Templated steps do not have any usable attributes
208			return {}
209

sportarchive / CloudProcessingEngine-Decider

Pull Request — master (#35)

ActivityStep.__init__() B

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing __init__.py files

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

ActivityStep.init() B

2. Missing init.py files

2. Missing init.py files