ThirdPartyResource.run() - Code Metrics - Inspection of "initial kubernetes pack with thirdpartyresource in..." - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#395)

unknown

created 2016-02-16 19:59 UTC

ThirdPartyResource.run() A

↳ Parent: ThirdPartyResource

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	12
rs	9.4285

#!/usr/bin/env python
import json
import requests
import sys
from requests.auth import HTTPBasicAuth
import ast
# from [email protected]:mward29/python-k8sclient.git

from st2reactor.sensor.base import Sensor


class ThirdPartyResource(Sensor):
    def setup(self):
        self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._logger.debug('Connecting to Kubernetes via api_client')
        extension = self._config['extension_url']
        KUBERNETES_API_URL = self._config['kubernetes_api_url'] + extension
        user = self._config['user']
        password = self._config['password']
#        verify = self._config['verify']
        self.client = requests.get(KUBERNETES_API_URL, auth=HTTPBasicAuth(user, password),
class Foo:
    def __init__(self, x=None):
        self.x = x
                                   verify=False, stream=True)

    def run(self):
        self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._logger.debug('Watch Kubernetes for thirdpartyresource information')
        r = self.client
        lines = r.iter_lines()
        # Save the first line for later or just skip it
#        first_line = next(lines)
        for line in lines:
            io = json.dumps(line)
            n = json.loads(io)
            d_list = ast.literal_eval(n)
            self._k8s_object(d_list=d_list)
#            self._resource_version(d_list=d_list)

    def _k8s_object(self, d_list):
        # Define some variables
        try:
            resource_type = d_list['type']
            object_kind = d_list['object']['kind']
            name = d_list['object']['metadata']['name']
            namespace = d_list['object']['metadata']['namespace']
            uid = d_list['object']['metadata']['uid']
        except:
            self.logger.debug("type, kind, name, namespace or uid do not exist in the object.\
                              must exit")
            sys.exit()
        # Now lets see if labels exist, if so build a trigger
        if 'labels' in d_list['object']['metadata']:
            labels_data = d_list['object']['metadata']['labels']
            self._build_a_trigger(resource_type=resource_type, name=name, labels=labels_data,
                                  namespace=namespace, object_kind=object_kind, uid=uid)
        else:
            self.logger.debug("No Labels for the resource below. Tough to proceed without knowing how \

                  to work with this object.")
            self.logger.debug(name, namespace, uid)

    def _build_a_trigger(self, resource_type, name, labels, namespace, object_kind, uid):
        trigger = 'kubernetes.thirdpartyobject'
        payload = {
            'resource': resource_type,
            'name': name,
            'labels': labels,
            'namespace': namespace,
            'object_kind': object_kind,
            'uid': uid
        }
        self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._logger.debug('Triggering Dispatch Now')

        # Create dispatch trigger
        self._sensor_service.dispatch(trigger=trigger, payload=payload)

    def cleanup(self):
        pass

    def add_trigger(self, trigger):
        pass

    def update_trigger(self, trigger):
        pass

    def remove_trigger(self, trigger):
        pass

    def _process_message(self, message):

        pass


1			#!/usr/bin/env python
2			import json
3			import requests
4			import sys
5			from requests.auth import HTTPBasicAuth
6			import ast
7			# from [email protected]:mward29/python-k8sclient.git
8
9			from st2reactor.sensor.base import Sensor
10
11
12			class ThirdPartyResource(Sensor):
13			def setup(self):
14			self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
			0 ignored issues – show Coding Style introduced 2016-02-16 20:00 UTC by Report Bug Copy Issue Report The attribute `_logger` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
15			self._logger.debug('Connecting to Kubernetes via api_client')
16			extension = self._config['extension_url']
17			KUBERNETES_API_URL = self._config['kubernetes_api_url'] + extension
18			user = self._config['user']
19			password = self._config['password']
20			# verify = self._config['verify']
21			self.client = requests.get(KUBERNETES_API_URL, auth=HTTPBasicAuth(user, password),
			0 ignored issues – show Coding Style introduced 2016-02-13 01:30 UTC by Report Bug Copy Issue Report The attribute `client` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
22			verify=False, stream=True)
23
24			def run(self):
25			self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
			0 ignored issues – show Coding Style introduced 2016-02-16 20:00 UTC by Report Bug Copy Issue Report The attribute `_logger` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
26			self._logger.debug('Watch Kubernetes for thirdpartyresource information')
27			r = self.client
28			lines = r.iter_lines()
29			# Save the first line for later or just skip it
30			# first_line = next(lines)
31			for line in lines:
32			io = json.dumps(line)
33			n = json.loads(io)
34			d_list = ast.literal_eval(n)
35			self._k8s_object(d_list=d_list)
36			# self._resource_version(d_list=d_list)
37
38			def _k8s_object(self, d_list):
39			# Define some variables
40			try:
41			resource_type = d_list['type']
42			object_kind = d_list['object']['kind']
43			name = d_list['object']['metadata']['name']
44			namespace = d_list['object']['metadata']['namespace']
45			uid = d_list['object']['metadata']['uid']
46			except:
47			self.logger.debug("type, kind, name, namespace or uid do not exist in the object.\
48			must exit")
49			sys.exit()
50			# Now lets see if labels exist, if so build a trigger
51			if 'labels' in d_list['object']['metadata']:
52			labels_data = d_list['object']['metadata']['labels']
53			self._build_a_trigger(resource_type=resource_type, name=name, labels=labels_data,
54			namespace=namespace, object_kind=object_kind, uid=uid)
55			else:
56			self.logger.debug("No Labels for the resource below. Tough to proceed without knowing how \
			0 ignored issues – show Coding Style introduced 2016-02-16 20:01 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (103/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
57			to work with this object.")
58			self.logger.debug(name, namespace, uid)
59
60			def _build_a_trigger(self, resource_type, name, labels, namespace, object_kind, uid):
61			trigger = 'kubernetes.thirdpartyobject'
62			payload = {
63			'resource': resource_type,
64			'name': name,
65			'labels': labels,
66			'namespace': namespace,
67			'object_kind': object_kind,
68			'uid': uid
69			}
70			self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)
			0 ignored issues – show Coding Style introduced 2016-02-16 20:00 UTC by Report Bug Copy Issue Report The attribute `_logger` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
71			self._logger.debug('Triggering Dispatch Now')
72
73			# Create dispatch trigger
74			self._sensor_service.dispatch(trigger=trigger, payload=payload)
75
76			def cleanup(self):
77			pass
78
79			def add_trigger(self, trigger):
80			pass
81
82			def update_trigger(self, trigger):
83			pass
84
85			def remove_trigger(self, trigger):
86			pass
87
88			def _process_message(self, message):
89
90			pass
91

StackStorm / st2contrib

Pull Request — master (#395)

ThirdPartyResource.run() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like