Activity.check_input() - Code Metrics - sportarchive/CloudProcessingEngine-Decider - Measure and Improve Code Quality continuously with Scrutinizer

Activity.check_input() A
last analyzed 2016-11-10 17:43 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	1
dl	0
loc	2
rs	10
c	1
b	1
f	0

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

import logging

from .schema import SchemaValidator

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

_LOGGER = logging.getLogger(__name__)


class Activity(object):
    """Activity abstraction.

    Attributes:
        name (str): The name of this `Activity`.
        version (str): The version of this `Activity`.
        task_list (str): Name of the  SWF `task_list` to use when submitting
                         activity tasks.
        heartbeat_timeout (str): SWF 'heartbeat_timeout' value.
        schedule_to_close_timeout (str): SWF 'schedule_to_close_timeout' value.
        schedule_to_start_timeout (str): SWF 'schedule_to_start_timeout' value.
        start_to_close_timeout (str): SWF 'start_to_close_timeout' value.

    Examples:
        >>> a = activity.Activity.from_data(
        ...     {
        ...         'name': 'MyActivity',
        ...         'version': '1.0',
        ...         'task_list': 'TranscodeAsset',
        ...         'heartbeat_timeout': '60',
        ...         'schedule_to_close_timeout': '518400',
        ...         'schedule_to_start_timeout': '43200',
        ...         'start_to_close_timeout': '432000'
        ...     }
        ... )
        >>> a
        Activity(name='MyActivity')
        >>> a.name
        'MyActivity'

    """

    _DATA_SCHEMA = {
        '$schema': 'http://json-schema.org/draft-04/schema#',
        'type': 'object',
        'properties': {
            'name': {
                'type': 'string',
            },
            'version': {
                'type': 'string',
            },
            'input_spec': {
                'oneOf': [
                    {'type': 'null'},
                    {'$ref': '#/definitions/input_spec'},
                ],
            },
            'outputs_spec': {
                'oneOf': [
                    {'type': 'null'},
                    {'$ref': '#/definitions/outputs_spec'},
                ]
            },
            'task_list': {
                'type': 'string',
            },
            'heartbeat_timeout': {
                'type': 'string',
            },
            'schedule_to_close_timeout': {
                'type': 'string',
            },
            'schedule_to_start_timeout': {
                'type': 'string',
            },
            'start_to_close_timeout': {
                'type': 'string',
            },
        },
        'additionalProperties': False,
        'definitions': {
            'input_spec': {
                '$ref': 'http://json-schema.org/draft-04/schema#',
            },
            'outputs_spec': {
                'type': 'object',
                'patternProperties': {
                    '^[a-zA-Z0-9]+$': {}
                },
                'minProperties': 1,
                'additionalProperties': False,
            },
        },
    }

    __slots__ = ('name', 'version',
                 'task_list',
                 'heartbeat_timeout',
                 'schedule_to_start_timeout',
                 'schedule_to_close_timeout',
                 'start_to_close_timeout',
                 '_input_validator', '_outputs_spec')

    def __init__(self, name, version,
                 input_spec=None, outputs_spec=None,
                 task_list=None,
                 heartbeat_timeout='60',
                 schedule_to_close_timeout='518400',
                 schedule_to_start_timeout='43200',
                 start_to_close_timeout='432000'):

        self.name = name
        self.version = version

        if task_list is None:
            task_list = '{name}-{version}'.format(
                name=self.name,
                version=self.version
            )
        self.task_list = task_list

        self.heartbeat_timeout = heartbeat_timeout
        self.schedule_to_close_timeout = schedule_to_close_timeout
        self.schedule_to_start_timeout = schedule_to_start_timeout
        self.start_to_close_timeout = start_to_close_timeout

        self._input_validator = SchemaValidator(input_spec)
        if outputs_spec:
            try:
                self._outputs_spec = {key: yaql.parse(expr)
                                      for key, expr in outputs_spec.items()}

            except Exception as err:
                _LOGGER.critical('Invalid YAQL expression in Activity %r: %r',
                                 name, err)
                raise
        else:
            self._outputs_spec = {}

    def __repr__(self):
        return 'Activity(name={name!r})'.format(name=self.name)

    def render_outputs(self, output):
        """Use the `Activity`'s `outputs_spec` to generate all the defined
        representation of this activity's output.
        """
        return {
            key: expr.evaluate(output)
            for key, expr in self._outputs_spec.items()
        }

    def check_input(self, activity_input):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        return self._input_validator.validate(activity_input)

    @classmethod
    def from_data(cls, data):
        """Define an `Activity` from a dictionary of attributes.
        """
        validator = SchemaValidator(cls._DATA_SCHEMA)
        validator.validate(data)

        activity_data = {
            'name': data['name'],
            'version': data['version'],
            'input_spec': data.get('input_spec', None),
            'outputs_spec': data.get('outputs_spec', None),
            'task_list': data.get('task_list', None)
        }

        # Copy in all SWF activity options.
        for option in ('heartbeat_timeout', 'schedule_to_close_timeout',
                       'schedule_to_start_timeout', 'start_to_close_timeout'):
            if option in data:
                activity_data[option] = data[option]
                

        return cls(**activity_data)



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 logging
8
9			from .schema import SchemaValidator
10
11			import yaql
			0 ignored issues – show Configuration introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report The import `yaql` 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
13			_LOGGER = logging.getLogger(__name__)
14
15
16			class Activity(object):
17			"""Activity abstraction.
18
19			Attributes:
20			name (str): The name of this `Activity`.
21			version (str): The version of this `Activity`.
22			task_list (str): Name of the SWF `task_list` to use when submitting
23			activity tasks.
24			heartbeat_timeout (str): SWF 'heartbeat_timeout' value.
25			schedule_to_close_timeout (str): SWF 'schedule_to_close_timeout' value.
26			schedule_to_start_timeout (str): SWF 'schedule_to_start_timeout' value.
27			start_to_close_timeout (str): SWF 'start_to_close_timeout' value.
28
29			Examples:
30			>>> a = activity.Activity.from_data(
31			... {
32			... 'name': 'MyActivity',
33			... 'version': '1.0',
34			... 'task_list': 'TranscodeAsset',
35			... 'heartbeat_timeout': '60',
36			... 'schedule_to_close_timeout': '518400',
37			... 'schedule_to_start_timeout': '43200',
38			... 'start_to_close_timeout': '432000'
39			... }
40			... )
41			>>> a
42			Activity(name='MyActivity')
43			>>> a.name
44			'MyActivity'
45
46			"""
47
48			_DATA_SCHEMA = {
49			'$schema': 'http://json-schema.org/draft-04/schema#',
50			'type': 'object',
51			'properties': {
52			'name': {
53			'type': 'string',
54			},
55			'version': {
56			'type': 'string',
57			},
58			'input_spec': {
59			'oneOf': [
60			{'type': 'null'},
61			{'$ref': '#/definitions/input_spec'},
62			],
63			},
64			'outputs_spec': {
65			'oneOf': [
66			{'type': 'null'},
67			{'$ref': '#/definitions/outputs_spec'},
68			]
69			},
70			'task_list': {
71			'type': 'string',
72			},
73			'heartbeat_timeout': {
74			'type': 'string',
75			},
76			'schedule_to_close_timeout': {
77			'type': 'string',
78			},
79			'schedule_to_start_timeout': {
80			'type': 'string',
81			},
82			'start_to_close_timeout': {
83			'type': 'string',
84			},
85			},
86			'additionalProperties': False,
87			'definitions': {
88			'input_spec': {
89			'$ref': 'http://json-schema.org/draft-04/schema#',
90			},
91			'outputs_spec': {
92			'type': 'object',
93			'patternProperties': {
94			'^[a-zA-Z0-9]+$': {}
95			},
96			'minProperties': 1,
97			'additionalProperties': False,
98			},
99			},
100			}
101
102			__slots__ = ('name', 'version',
103			'task_list',
104			'heartbeat_timeout',
105			'schedule_to_start_timeout',
106			'schedule_to_close_timeout',
107			'start_to_close_timeout',
108			'_input_validator', '_outputs_spec')
109
110			def __init__(self, name, version,
111			input_spec=None, outputs_spec=None,
112			task_list=None,
113			heartbeat_timeout='60',
114			schedule_to_close_timeout='518400',
115			schedule_to_start_timeout='43200',
116			start_to_close_timeout='432000'):
117
118			self.name = name
119			self.version = version
120
121			if task_list is None:
122			task_list = '{name}-{version}'.format(
123			name=self.name,
124			version=self.version
125			)
126			self.task_list = task_list
127
128			self.heartbeat_timeout = heartbeat_timeout
129			self.schedule_to_close_timeout = schedule_to_close_timeout
130			self.schedule_to_start_timeout = schedule_to_start_timeout
131			self.start_to_close_timeout = start_to_close_timeout
132
133			self._input_validator = SchemaValidator(input_spec)
134			if outputs_spec:
135			try:
136			self._outputs_spec = {key: yaql.parse(expr)
137			for key, expr in outputs_spec.items()}
138
139			except Exception as err:
140			_LOGGER.critical('Invalid YAQL expression in Activity %r: %r',
141			name, err)
142			raise
143			else:
144			self._outputs_spec = {}
145
146			def __repr__(self):
147			return 'Activity(name={name!r})'.format(name=self.name)
148
149			def render_outputs(self, output):
150			"""Use the `Activity`'s `outputs_spec` to generate all the defined
151			representation of this activity's output.
152			"""
153			return {
154			key: expr.evaluate(output)
155			for key, expr in self._outputs_spec.items()
156			}
157
158			def check_input(self, activity_input):
			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...
159			return self._input_validator.validate(activity_input)
160
161			@classmethod
162			def from_data(cls, data):
163			"""Define an `Activity` from a dictionary of attributes.
164			"""
165			validator = SchemaValidator(cls._DATA_SCHEMA)
166			validator.validate(data)
167
168			activity_data = {
169			'name': data['name'],
170			'version': data['version'],
171			'input_spec': data.get('input_spec', None),
172			'outputs_spec': data.get('outputs_spec', None),
173			'task_list': data.get('task_list', None)
174			}
175
176			# Copy in all SWF activity options.
177			for option in ('heartbeat_timeout', 'schedule_to_close_timeout',
178			'schedule_to_start_timeout', 'start_to_close_timeout'):
179			if option in data:
180			activity_data[option] = data[option]
181
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
182			return cls(**activity_data)
			0 ignored issues – show Coding Style introduced 2015-12-30 07:05 UTC by Report Bug Copy Issue Report Usage of `` or `` arguments should usually be done with care. Generally, there is nothing wrong with usage of `` or `**` arguments. For readability of the code base, we suggest to not over-use these language constructs though. For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect. Loading history...
183

sportarchive / CloudProcessingEngine-Decider

Activity.check_input() A last analyzed 2016-11-10 17:43 UTC

Complexity

Size

Duplication

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

Activity.check_input() A
last analyzed 2016-11-10 17:43 UTC

2. Missing init.py files