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

Plan.from_data() B
last analyzed 2016-11-10 17:43 UTC

↳ Parent: Plan

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	3
c	2
b	0
f	0
dl	0
loc	31
rs	8.8571

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

import logging

from .step import Step
from .activity import Activity
from .schema import SchemaValidator

_LOGGER = logging.getLogger(__name__)


class Plan(object):
    """Workflow plan.
    """

    _DATA_SCHEMA = {
        '$schema': 'http://json-schema.org/draft-04/schema#',
        'type': 'object',
        'properties': {
            'name': {
                'type': 'string',
            },
            'version': {
                'type': 'string',
            },
            'default_execution_start_to_close_timeout': {
                'type': 'string',
            },
            'default_task_start_to_close_timeout': {
                'type': 'string',
            },
            'input_spec': {
                'oneOf': [
                    {'type': 'null'},
                    {'$ref': '#/definitions/input_spec'},
                ],
            },
            'activities': {
                'type': 'array',
                'minItem': 1,
                'items': {
                    'type': 'object'
                }
            },
            'steps': {
                'type': 'array',
                'minItem': 1,
                'items': {
                    'type': 'object'
                }
            }
        },
        'additionalProperties': False,
        'definitions': {
            'input_spec': {
                '$ref': 'http://json-schema.org/draft-04/schema#',
            },
        },
    }

    __slots__ = ('name',
                 'version',
                 'default_execution_start_to_close_timeout',
                 'default_task_start_to_close_timeout',
                 'steps',
                 'activities',
                 '_input_validator',
                 '__weakref__')

    def __init__(self, name, version,
                 default_execution_start_to_close_timeout,
                 default_task_start_to_close_timeout,
                 input_spec=None, steps=(), activities=()):
        

        self.name = name
        self.version = version
        self.default_execution_start_to_close_timeout = default_execution_start_to_close_timeout

        self.default_task_start_to_close_timeout = default_task_start_to_close_timeout

        self.steps = list(steps)
        self.activities = dict(activities)
        self._input_validator = SchemaValidator(input_spec=input_spec)

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

    @classmethod
    def from_data(cls, plan_data):
        """Define a plan from a dictionary of attributes.
        """
        validator = SchemaValidator(cls._DATA_SCHEMA)
        validator.validate(plan_data)
        

        activities = {
            activity_data['name']: Activity.from_data(activity_data)
            for activity_data in plan_data['activities']
        }

        steps = []
        for step_data in plan_data['steps']:
            step = Step.from_data(step_data, activities)
            steps.append(step)

        plan = cls(
            name=plan_data['name'],
            version=plan_data['version'],
            default_execution_start_to_close_timeout=plan_data['default_execution_start_to_close_timeout'],

            default_task_start_to_close_timeout=plan_data['default_task_start_to_close_timeout'],

            input_spec=plan_data.get('input_spec', None),
            steps=steps,
            activities=activities,
        )

        _LOGGER.info('Loaded plan %s(steps:%d activities:%d)',
                      plan, len(steps), len(activities))

        return plan

    def __repr__(self):
        return 'Plan(name={name})'.format(name=self.name)


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 .step import Step
10			from .activity import Activity
11			from .schema import SchemaValidator
12
13			_LOGGER = logging.getLogger(__name__)
14
15
16			class Plan(object):
17			"""Workflow plan.
18			"""
19
20			_DATA_SCHEMA = {
21			'$schema': 'http://json-schema.org/draft-04/schema#',
22			'type': 'object',
23			'properties': {
24			'name': {
25			'type': 'string',
26			},
27			'version': {
28			'type': 'string',
29			},
30			'default_execution_start_to_close_timeout': {
31			'type': 'string',
32			},
33			'default_task_start_to_close_timeout': {
34			'type': 'string',
35			},
36			'input_spec': {
37			'oneOf': [
38			{'type': 'null'},
39			{'$ref': '#/definitions/input_spec'},
40			],
41			},
42			'activities': {
43			'type': 'array',
44			'minItem': 1,
45			'items': {
46			'type': 'object'
47			}
48			},
49			'steps': {
50			'type': 'array',
51			'minItem': 1,
52			'items': {
53			'type': 'object'
54			}
55			}
56			},
57			'additionalProperties': False,
58			'definitions': {
59			'input_spec': {
60			'$ref': 'http://json-schema.org/draft-04/schema#',
61			},
62			},
63			}
64
65			__slots__ = ('name',
66			'version',
67			'default_execution_start_to_close_timeout',
68			'default_task_start_to_close_timeout',
69			'steps',
70			'activities',
71			'_input_validator',
72			'__weakref__')
73
74			def __init__(self, name, version,
75			default_execution_start_to_close_timeout,
76			default_task_start_to_close_timeout,
77			input_spec=None, steps=(), activities=()):
78
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
79			self.name = name
80			self.version = version
81			self.default_execution_start_to_close_timeout = default_execution_start_to_close_timeout
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (96/80). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Coding Style Naming introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report The name `default_execution_start_to_close_timeout` does not conform to the attribute naming conventions (`[a-z_][a-z0-9_]{2,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
82			self.default_task_start_to_close_timeout = default_task_start_to_close_timeout
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (86/80). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Coding Style Naming introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report The name `default_task_start_to_close_timeout` does not conform to the attribute naming conventions (`[a-z_][a-z0-9_]{2,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
83			self.steps = list(steps)
84			self.activities = dict(activities)
85			self._input_validator = SchemaValidator(input_spec=input_spec)
86
87			def check_input(self, plan_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...
88			return self._input_validator.validate(plan_input)
89
90			@classmethod
91			def from_data(cls, plan_data):
92			"""Define a plan from a dictionary of attributes.
93			"""
94			validator = SchemaValidator(cls._DATA_SCHEMA)
95			validator.validate(plan_data)
96
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
97			activities = {
98			activity_data['name']: Activity.from_data(activity_data)
99			for activity_data in plan_data['activities']
100			}
101
102			steps = []
103			for step_data in plan_data['steps']:
104			step = Step.from_data(step_data, activities)
105			steps.append(step)
106
107			plan = cls(
108			name=plan_data['name'],
109			version=plan_data['version'],
110			default_execution_start_to_close_timeout=plan_data['default_execution_start_to_close_timeout'],
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (107/80). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
111			default_task_start_to_close_timeout=plan_data['default_task_start_to_close_timeout'],
			0 ignored issues – show Coding Style introduced 2016-11-10 17:43 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (97/80). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
112			input_spec=plan_data.get('input_spec', None),
113			steps=steps,
114			activities=activities,
115			)
116
117			_LOGGER.info('Loaded plan %s(steps:%d activities:%d)',
118			plan, len(steps), len(activities))
119
120			return plan
121
122			def __repr__(self):
123			return 'Plan(name={name})'.format(name=self.name)
124

sportarchive / CloudProcessingEngine-Decider

Plan.from_data() B last analyzed 2016-11-10 17:43 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Plan.from_data() B
last analyzed 2016-11-10 17:43 UTC