ThirdPartyResource

Complexity

Total Complexity

12

Size/Duplication

Total Lines	80
Duplicated Lines	0 %

10 Methods

Rating	Name	Duplication	Size	Complexity
A	remove_trigger()	0	2	1
A	run()	0	14	2
A	cleanup()	0	2	1
A	setup()	0	9	1
A	_build_a_trigger()	0	16	1
A	__init__()	0	3	1
A	add_trigger()	0	2	1
A	_k8s_object()	0	16	2
A	update_trigger()	0	2	1
A	_process_message()	0	3	1

#!/usr/bin/env python
import base64

The import base64 seems to be unused.

import json
import requests
from requests.auth import HTTPBasicAuth
import yaml

The import yaml seems to be unused.

import re

The import re seems to be unused.

import os, sys

The import os seems to be unused.

The import sys seems to be unused.

import pprint

The import pprint seems to be unused.

import ast
import uuid

The import uuid seems to be unused.

# from git@github.com:mward29/python-k8sclient.git

from st2reactor.sensor.base import Sensor


class ThirdPartyResource(Sensor):
    def __init__(self, sensor_service, config=None):
        super(ThirdPartyResource, self).__init__(sensor_service=sensor_service,
                                                  config=config)
#        self._labels = self._config['labels'].get('thirdpartyresource', [])\

    def setup(self):
        self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)

The attribute _logger was defined outside __init__.

        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']
        self.client = requests.get(KUBERNETES_API_URL, auth=HTTPBasicAuth(user, password), verify=False, stream=True)

This line is too long as per the coding-style (117/100).

The attribute client was defined outside __init__.

        pass

Unnecessary pass statement

    def run(self):
        self._logger = self._sensor_service.get_logger(name=self.__class__.__name__)

The attribute _logger was defined outside __init__.

        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)

The variable first_line seems to be unused.

        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)
        pass

Unnecessary pass statement

    def _k8s_object(self, d_list):
        # Define some variables
        resource_type = d_list['type']
        object_kind = d_list['object']['kind']
        api_version = d_list['object']['apiVersion']

The variable api_version seems to be unused.

        name = d_list['object']['metadata']['name']
        namespace = d_list['object']['metadata']['namespace']
        uid = d_list['object']['metadata']['uid']
        # 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)

This line is too long as per the coding-style (148/100).

        else:
            print("No Labels for the resource below. Tough to proceed without knowing how to work with this object.")

This line is too long as per the coding-style (117/100).

            print(name, namespace, uid)
        pass

Unnecessary pass statement

    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__)

The attribute _logger was defined outside __init__.

        self._logger.debug('Triggering Dispatch Now')

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

Unnecessary pass statement

    def cleanup(self):
        pass

    def add_trigger(self,  trigger):

Exactly one space required after comma
def add_trigger(self, trigger):
^

        pass

    def update_trigger(self, trigger):
        pass

    def remove_trigger(self, trigger):
        pass

    def _process_message(self, message):

        pass