ThirdPartyResource.run() - Code Metrics - Inspection of "WIP: Improvements to kubernetes pack!" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#400)

unknown

created 2016-02-17 20:22 UTC

ThirdPartyResource.run() B

↳ Parent: ThirdPartyResource

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	5
dl	0
loc	20
rs	8.5454

#!/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._log = self._sensor_service.get_logger(__name__)
class Foo:
    def __init__(self, x=None):
        self.x = x
        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._log.debug('Connecting to Kubernetes endpoint %s via api_client.' %
                        KUBERNETES_API_URL)
        self.client = requests.get(KUBERNETES_API_URL, auth=HTTPBasicAuth(user, password),
class Foo:
    def __init__(self, x=None):
        self.x = x
                                   verify=verify, stream=True)

    def run(self):
        self._log.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)

        trigger = 'kubernetes.thirdpartyobject'

        for line in lines:
            try:
                trigger_payload = self._get_trigger_payload_from_line(line)
            except:
                msg = ('Failed generating trigger payload from line %s. Aborting sensor!!!' %
                    line)
                self._log.exception(msg)
                sys.exit(1)
            else:
                self._log.debug('Triggering Dispatch Now')
                self._sensor_service.dispatch(trigger=trigger, payload=trigger_payload)

    def _get_trigger_payload_from_line(self, line):
        # need to perform a json dump due to uft8 error prior to performing a json.load
        io = json.dumps(line)
        n = json.loads(io)
        line = ast.literal_eval(n)
        payload = self._k8s_object(line)
        return payload

    def _k8s_object(self, line):
        # Define some variables
        try:
            resource_type = line['type']
            object_kind = line['object']['kind']
            name = line['object']['metadata']['name']
            namespace = line['object']['metadata']['namespace']
            uid = line['object']['metadata']['uid']
        except KeyError:
            msg = 'One of "type", "kind", "name", "namespace" or "uid" ' + \
                  'do not exist in the object.'
            self._log.exception(msg)
            raise

        # Now lets see if labels exist, if so build a trigger else exit
        if 'labels' in line['object']['metadata']:
            labels_data = line['object']['metadata']['labels']
            payload = self._build_a_trigger(resource_type=resource_type, name=name,
                                            labels=labels_data, namespace=namespace,
                                            object_kind=object_kind, uid=uid)
            self._log.debug('Trigger payload: %s.' % payload)
            return payload
        else:
            msg = 'No Labels for the resource below. Tough to proceed without knowing how ' + \
                  'to work with this object. Incoming object=%s' % line
            self._log.error(msg)
            raise Exception(msg)

    def _build_a_trigger(self, resource_type, name, labels, namespace, object_kind, uid):
        payload = {
            'resource': resource_type,
            'name': name,
            'labels': labels,
            'namespace': namespace,
            'object_kind': object_kind,
            'uid': uid
        }

        return payload


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._log = self._sensor_service.get_logger(__name__)
			0 ignored issues – show Coding Style introduced 2016-02-16 21:24 UTC by Report Bug Copy Issue Report The attribute `_log` 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			extension = self._config['extension_url']
16			KUBERNETES_API_URL = self._config['kubernetes_api_url'] + extension
17			user = self._config['user']
18			password = self._config['password']
19			verify = self._config['verify']
20			self._log.debug('Connecting to Kubernetes endpoint %s via api_client.' %
21			KUBERNETES_API_URL)
22			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...
23			verify=verify, stream=True)
24
25			def run(self):
26			self._log.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
32			trigger = 'kubernetes.thirdpartyobject'
33
34			for line in lines:
35			try:
36			trigger_payload = self._get_trigger_payload_from_line(line)
37			except:
38			msg = ('Failed generating trigger payload from line %s. Aborting sensor!!!' %
39			line)
40			self._log.exception(msg)
41			sys.exit(1)
42			else:
43			self._log.debug('Triggering Dispatch Now')
44			self._sensor_service.dispatch(trigger=trigger, payload=trigger_payload)
45
46			def _get_trigger_payload_from_line(self, line):
47			# need to perform a json dump due to uft8 error prior to performing a json.load
48			io = json.dumps(line)
49			n = json.loads(io)
50			line = ast.literal_eval(n)
51			payload = self._k8s_object(line)
52			return payload
53
54			def _k8s_object(self, line):
55			# Define some variables
56			try:
57			resource_type = line['type']
58			object_kind = line['object']['kind']
59			name = line['object']['metadata']['name']
60			namespace = line['object']['metadata']['namespace']
61			uid = line['object']['metadata']['uid']
62			except KeyError:
63			msg = 'One of "type", "kind", "name", "namespace" or "uid" ' + \
64			'do not exist in the object.'
65			self._log.exception(msg)
66			raise
67
68			# Now lets see if labels exist, if so build a trigger else exit
69			if 'labels' in line['object']['metadata']:
70			labels_data = line['object']['metadata']['labels']
71			payload = self._build_a_trigger(resource_type=resource_type, name=name,
72			labels=labels_data, namespace=namespace,
73			object_kind=object_kind, uid=uid)
74			self._log.debug('Trigger payload: %s.' % payload)
75			return payload
76			else:
77			msg = 'No Labels for the resource below. Tough to proceed without knowing how ' + \
78			'to work with this object. Incoming object=%s' % line
79			self._log.error(msg)
80			raise Exception(msg)
81
82			def _build_a_trigger(self, resource_type, name, labels, namespace, object_kind, uid):
83			payload = {
84			'resource': resource_type,
85			'name': name,
86			'labels': labels,
87			'namespace': namespace,
88			'object_kind': object_kind,
89			'uid': uid
90			}
91
92			return payload
93

StackStorm / st2contrib

Pull Request — master (#400)

ThirdPartyResource.run() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like