Events - Code Metrics - Inspection of "Created sample email rule" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#544)

unknown

created 2016-08-17 13:05 UTC

Events A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	30
Duplicated Lines	0 %

Importance

Changes

Metric	Value
dl	0
loc	30
rs	10
c	0
b	0
f	0
wmc	6

import base64
import json
import os
import requests

import yaml


def parseOutput(r):
    try:
        output = {}
        json.loads(r.text)
        for c in r.json():
            output[c['name']] = c
        return json.dumps(output)
    except:
        return r.text


class Sensu(object):

    def __init__(self, conf):

        config_file = os.path.join(os.path.dirname(__file__), conf)
        try:
            fh = open(config_file)
            self.config = yaml.safe_load(fh)
            fh.close()
        except Exception as e:
            print("Error reading config file %s: %s" % (conf, e))

        if self.config['ssl']:
            protocol = 'https'
        else:
            protocol = "http"
        self.config[
            'base_url'] = "%s://%s:%s" % (protocol, self.config['host'], self.config['port'])

    def get_headers(self):
        b64auth = base64.b64encode(
            "%s:%s" %
            (self.config['user'], self.config['pass']))
        auth_header = "BASIC %s" % b64auth
        content_header = "application/json"
        return {"Authorization": auth_header, "Content-Type": content_header}


class Aggregates(object):

    def __init__(self, conf):
        sensu = Sensu(conf)
        self.headers = sensu.get_headers()
        self.url = "%s/aggregates" % sensu.config['base_url']

    def list(self, limit=None, offset=None):
        data = {}
        if limit:
            data['limit'] = limit
        if offset:
            data['offset'] = offset

        return parseOutput(
            requests.get(
                url=self.url,
                headers=self.headers,
                data=data))

    def check(self, check, age=None):
        data = {}
        self.url = "%s/%s" % (self.url, check)
        if age:
            data['age'] = age
        return parseOutput(
            requests.get(
                url=self.url,
                headers=self.headers,
                data=data))

    def delete(self, check):
        self.url = "%s/%s" % (self.url, check)
        return parseOutput(requests.delete(url=self.url, headers=self.headers))

    def check_issued(self, check, issued, summarize=None, results=None):
        data = {}
        self.url = "%s/%s/%s" % (self.url, check, issued)
        if summarize:
            data['summarize'] = summarize
        if results:
            data['results'] = results
        return parseOutput(
            requests.get(
                url=self.url,
                headers=self.headers,
                data=data))


class Checks(object):

    def __init__(self, conf):
        self.sensu = Sensu(conf)
        self.headers = self.sensu.get_headers()
        self.url = "%s/checks" % self.sensu.config['base_url']

    def list(self):
        return parseOutput(requests.get(url=self.url, headers=self.headers))

    def get(self, check):
        url = "%s/%s" % (self.url, check)
        return parseOutput(requests.get(url=url, headers=self.headers))

    def request(self, check, subscribers):
        url = "%s/request" % self.sensu.config['base_url']
        payload = {}
        subs = []
        if not isinstance(subscribers, list):
            subs.append(subscribers)
        payload['check'] = check
        payload['subscribers'] = subscribers
        return parseOutput(
            requests.post(
                url=url,
                headers=self.headers,
                data=payload))


class Clients(object):

    def __init__(self, conf):
        sensu = Sensu(conf)
        self.headers = sensu.get_headers()
        self.url = "%s/clients" % sensu.config['base_url']

    def list(self, limit=None, offset=None):
        data = {}
        if limit:
            data['limit'] = limit
        if offset:
            data['offset'] = offset

        return parseOutput(
            requests.get(
                url=self.url,
                headers=self.headers,
                data=data))

    def get(self, client):
        url = "%s/%s" % (self.url, client)
        return parseOutput(requests.get(url=url, headers=self.headers))

    def delete(self, client):
        url = "%s/%s" % (self.url, client)
        return parseOutput(requests.delete(url=url, headers=self.headers))

    def history(self, client):
        url = "%s/%s/history" % (self.url, client)
        return parseOutput(requests.get(url=url, headers=self.headers))


class Stashes(object):

    def __init__(self, conf):
        sensu = Sensu(conf)
        self.headers = sensu.get_headers()
        self.url = "%s/stashes" % sensu.config['base_url']

    def list(self, limit, offset):
        data = {}
        if limit:
            data['limit'] = limit
        if offset:
            data['offset'] = offset
        return parseOutput(
            requests.get(
                url=self.url,
                headers=self.headers,
                data=data))

    def get(self, stash):
        url = "%s/%s" % (self.url, stash)
        return parseOutput(requests.get(url=url, headers=self.headers))

    def delete(self, stash):
        url = "%s/%s" % (self.url, stash)
        return parseOutput(requests.delete(url=url, headers=self.headers))

    def post(self, data):
        return parseOutput(
            requests.post(
                url=self.url,
                headers=self.headers,
                data=data))

    def post_by_path(self, path, data):
        url = "%s/%s" % (self.url, path)
        return parseOutput(
            requests.post(
                url=url,
                headers=self.headers,
                data=data))


class Status(object):

    def __init__(self, conf):
        sensu = Sensu(conf)
        self.headers = sensu.get_headers()
        self.url = sensu.config['base_url']

    def health(self, consumers=2, messages=100):
        url = "%s/health" % self.url
        data = {'consumers': consumers, 'messages': messages}
        return parseOutput(
            requests.get(
                url=url,
                headers=self.headers,
                data=data))

    def info(self):
        url = "%s/info" % self.url
        return parseOutput(requests.get(url=url, headers=self.headers))


class Events(object):

    def __init__(self, conf):
        self.sensu = Sensu(conf)
        self.headers = self.sensu.get_headers()
        self.url = "%s/events" % self.sensu.config['base_url']

    def list_all(self):
        return parseOutput(requests.get(url=self.url, headers=self.headers))

    def list_by_client(self, client):
        url = "%s/%s" % (self.url, client)
        return parseOutput(requests.get(url=url, headers=self.headers))

    def get(self, client, check):
        url = "%s/%s/%s" % (self.url, client, check)
        return parseOutput(requests.get(url=url, headers=self.headers))

    def delete(self, client, check):
        url = "%s/%s/%s" % (self.url, client, check)
        return parseOutput(requests.delete(url=url, headers=self.headers))

    def resolve(self, client, check):
        url = "%s/resolve" % self.sensu['base_url']
        payload = {'client': client, 'check': check}
        return parseOutput(
            requests.post(
                url=url,
                headers=self.headers,
                data=payload))


1			import base64
2			import json
3			import os
4			import requests
5
6			import yaml
7
8
9			def parseOutput(r):
10			try:
11			output = {}
12			json.loads(r.text)
13			for c in r.json():
14			output[c['name']] = c
15			return json.dumps(output)
16			except:
17			return r.text
18
19
20			class Sensu(object):
21
22			def __init__(self, conf):
23
24			config_file = os.path.join(os.path.dirname(__file__), conf)
25			try:
26			fh = open(config_file)
27			self.config = yaml.safe_load(fh)
28			fh.close()
29			except Exception as e:
30			print("Error reading config file %s: %s" % (conf, e))
31
32			if self.config['ssl']:
33			protocol = 'https'
34			else:
35			protocol = "http"
36			self.config[
37			'base_url'] = "%s://%s:%s" % (protocol, self.config['host'], self.config['port'])
38
39			def get_headers(self):
40			b64auth = base64.b64encode(
41			"%s:%s" %
42			(self.config['user'], self.config['pass']))
43			auth_header = "BASIC %s" % b64auth
44			content_header = "application/json"
45			return {"Authorization": auth_header, "Content-Type": content_header}
46
47
48			class Aggregates(object):
49
50			def __init__(self, conf):
51			sensu = Sensu(conf)
52			self.headers = sensu.get_headers()
53			self.url = "%s/aggregates" % sensu.config['base_url']
54
55			def list(self, limit=None, offset=None):
56			data = {}
57			if limit:
58			data['limit'] = limit
59			if offset:
60			data['offset'] = offset
61
62			return parseOutput(
63			requests.get(
64			url=self.url,
65			headers=self.headers,
66			data=data))
67
68			def check(self, check, age=None):
69			data = {}
70			self.url = "%s/%s" % (self.url, check)
71			if age:
72			data['age'] = age
73			return parseOutput(
74			requests.get(
75			url=self.url,
76			headers=self.headers,
77			data=data))
78
79			def delete(self, check):
80			self.url = "%s/%s" % (self.url, check)
81			return parseOutput(requests.delete(url=self.url, headers=self.headers))
82
83			def check_issued(self, check, issued, summarize=None, results=None):
84			data = {}
85			self.url = "%s/%s/%s" % (self.url, check, issued)
86			if summarize:
87			data['summarize'] = summarize
88			if results:
89			data['results'] = results
90			return parseOutput(
91			requests.get(
92			url=self.url,
93			headers=self.headers,
94			data=data))
95
96
97			class Checks(object):
98
99			def __init__(self, conf):
100			self.sensu = Sensu(conf)
101			self.headers = self.sensu.get_headers()
102			self.url = "%s/checks" % self.sensu.config['base_url']
103
104			def list(self):
105			return parseOutput(requests.get(url=self.url, headers=self.headers))
106
107			def get(self, check):
108			url = "%s/%s" % (self.url, check)
109			return parseOutput(requests.get(url=url, headers=self.headers))
110
111			def request(self, check, subscribers):
112			url = "%s/request" % self.sensu.config['base_url']
113			payload = {}
114			subs = []
115			if not isinstance(subscribers, list):
116			subs.append(subscribers)
117			payload['check'] = check
118			payload['subscribers'] = subscribers
119			return parseOutput(
120			requests.post(
121			url=url,
122			headers=self.headers,
123			data=payload))
124
125
126			class Clients(object):
127
128			def __init__(self, conf):
129			sensu = Sensu(conf)
130			self.headers = sensu.get_headers()
131			self.url = "%s/clients" % sensu.config['base_url']
132
133			def list(self, limit=None, offset=None):
134			data = {}
135			if limit:
136			data['limit'] = limit
137			if offset:
138			data['offset'] = offset
139
140			return parseOutput(
141			requests.get(
142			url=self.url,
143			headers=self.headers,
144			data=data))
145
146			def get(self, client):
147			url = "%s/%s" % (self.url, client)
148			return parseOutput(requests.get(url=url, headers=self.headers))
149
150			def delete(self, client):
151			url = "%s/%s" % (self.url, client)
152			return parseOutput(requests.delete(url=url, headers=self.headers))
153
154			def history(self, client):
155			url = "%s/%s/history" % (self.url, client)
156			return parseOutput(requests.get(url=url, headers=self.headers))
157
158
159			class Stashes(object):
160
161			def __init__(self, conf):
162			sensu = Sensu(conf)
163			self.headers = sensu.get_headers()
164			self.url = "%s/stashes" % sensu.config['base_url']
165
166			def list(self, limit, offset):
167			data = {}
168			if limit:
169			data['limit'] = limit
170			if offset:
171			data['offset'] = offset
172			return parseOutput(
173			requests.get(
174			url=self.url,
175			headers=self.headers,
176			data=data))
177
178			def get(self, stash):
179			url = "%s/%s" % (self.url, stash)
180			return parseOutput(requests.get(url=url, headers=self.headers))
181
182			def delete(self, stash):
183			url = "%s/%s" % (self.url, stash)
184			return parseOutput(requests.delete(url=url, headers=self.headers))
185
186			def post(self, data):
187			return parseOutput(
188			requests.post(
189			url=self.url,
190			headers=self.headers,
191			data=data))
192
193			def post_by_path(self, path, data):
194			url = "%s/%s" % (self.url, path)
195			return parseOutput(
196			requests.post(
197			url=url,
198			headers=self.headers,
199			data=data))
200
201
202			class Status(object):
203
204			def __init__(self, conf):
205			sensu = Sensu(conf)
206			self.headers = sensu.get_headers()
207			self.url = sensu.config['base_url']
208
209			def health(self, consumers=2, messages=100):
210			url = "%s/health" % self.url
211			data = {'consumers': consumers, 'messages': messages}
212			return parseOutput(
213			requests.get(
214			url=url,
215			headers=self.headers,
216			data=data))
217
218			def info(self):
219			url = "%s/info" % self.url
220			return parseOutput(requests.get(url=url, headers=self.headers))
221
222
223			class Events(object):
224
225			def __init__(self, conf):
226			self.sensu = Sensu(conf)
227			self.headers = self.sensu.get_headers()
228			self.url = "%s/events" % self.sensu.config['base_url']
229
230			def list_all(self):
231			return parseOutput(requests.get(url=self.url, headers=self.headers))
232
233			def list_by_client(self, client):
234			url = "%s/%s" % (self.url, client)
235			return parseOutput(requests.get(url=url, headers=self.headers))
236
237			def get(self, client, check):
238			url = "%s/%s/%s" % (self.url, client, check)
239			return parseOutput(requests.get(url=url, headers=self.headers))
240
241			def delete(self, client, check):
242			url = "%s/%s/%s" % (self.url, client, check)
243			return parseOutput(requests.delete(url=url, headers=self.headers))
244
245			def resolve(self, client, check):
246			url = "%s/resolve" % self.sensu['base_url']
247			payload = {'client': client, 'check': check}
248			return parseOutput(
249			requests.post(
250			url=url,
251			headers=self.headers,
252			data=payload))
253

StackStorm / st2contrib

Pull Request — master (#544)

Events A

Complexity

Size/Duplication

Importance

Duplication Side-by-Side

Filter issues like