gsmmodem.GSMModemCollection.__init__() - Code Metrics - Inspection of "Merge branch 'develop'" - myems/myems-api - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e07a89...15a904 )

by Guangyu

created 2020-11-26 09:23 UTC

gsmmodem.GSMModemCollection.init() A

↳ Parent: gsmmodem

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	3
Ratio	100 %

Importance

Changes

Metric	Value
eloc	3
dl	3
loc	3
rs	10
c	0
b	0
f	0
cc	1
nop	0

import falcon
import json
import mysql.connector
import config
import base64
import re


class GSMModemCollection:

    @staticmethod
    def __init__():
        pass

    @staticmethod
    def on_options(req, resp):
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        query = (" SELECT id, serial_port, baud_rate "
                 " FROM tbl_gsm_modems ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.disconnect()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0],
                               "serial_port": row[1],
                               "baud_rate": row[2]}
                result.append(meta_result)

        resp.body = json.dumps(result)

    @staticmethod
    def on_post(req, resp):
        """Handles POST requests"""
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR', description=ex)

        new_values = json.loads(raw_json, encoding='utf-8')

        if 'serial_port' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['serial_port'], str) or \
                len(str.strip(new_values['data']['serial_port'])) == 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SERIAL_PORT')

        serial_port = str.strip(new_values['data']['serial_port'])

        if 'baud_rate' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['baud_rate'], int) or \
                new_values['data']['baud_rate'] <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_BAUD_RATE')
        baud_rate = float(new_values['data']['baud_rate'])

        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        cursor.execute(" SELECT id "
                       " FROM tbl_gsm_modems "
                       " WHERE serial_port = %s ", (serial_port,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
                                   description='API.GSM_MODEM_SERIAL_PORT_IS_ALREADY_IN_USE')

        add_value = (" INSERT INTO tbl_gsm_modems "
                     "    (serial_port, baud_rate) "
                     " VALUES (%s, %s) ")
        cursor.execute(add_value, (serial_port,
                                   baud_rate))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.disconnect()

        resp.status = falcon.HTTP_201
        resp.location = '/gsmmodems/' + str(new_id)


class GSMModemItem:
    @staticmethod
    def __init__():
        pass

    @staticmethod
    def on_options(req, resp, id_):
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp, id_):
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, '400 Bad Request')

        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        query = (" SELECT id, serial_port, baud_rate "
                 " FROM tbl_gsm_modems "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.disconnect()
        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, 'API.NOT_FOUND')

        result = {"id": row[0],
                  "serial_port": row[1],
                  "baud_rate": row[2]}
        resp.body = json.dumps(result)

    @staticmethod

    def on_delete(req, resp, id_):
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_GSM_MODEM_ID')

        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        cursor.execute(" SELECT serial_port "
                       " FROM tbl_gsm_modems "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.GSM_MODEM_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_gsm_modems WHERE id = %s ", (id_,))
        cnx.commit()

        cursor.close()
        cnx.disconnect()

        resp.status = falcon.HTTP_204

    @staticmethod

    def on_put(req, resp, id_):
        """Handles PUT requests"""
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.EXCEPTION', description=ex)

        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_GSM_MODEM_ID')

        new_values = json.loads(raw_json, encoding='utf-8')

        if 'serial_port' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['serial_port'], str) or \
                len(str.strip(new_values['data']['serial_port'])) == 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SERIAL_PORT')

        serial_port = str.strip(new_values['data']['serial_port'])

        if 'baud_rate' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['baud_rate'], int) or \
                new_values['data']['baud_rate'] <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_BAUD_RATE')
        baud_rate = float(new_values['data']['baud_rate'])

        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        cursor.execute(" SELECT serial_port "
                       " FROM tbl_gsm_modems "
                       " WHERE id = %s ",
                       (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.GSM_MODEM_NOT_FOUND')

        cursor.execute(" SELECT serial_port "
                       " FROM tbl_gsm_modems "
                       " WHERE serial_port = %s AND id != %s ", (serial_port, id_))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
                                   description='API.GSM_MODEM_SERIAL_PORT_IS_ALREADY_IN_USE')

        update_row = (" UPDATE tbl_gsm_modems "
                      " SET serial_port = %s, baud_rate = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (serial_port,
                                    baud_rate,
                                    id_,))
        cnx.commit()

        cursor.close()
        cnx.disconnect()

        resp.status = falcon.HTTP_200


1		import falcon
2		import json
3		import mysql.connector
4		import config
5		import base64
6		import re
7
8
9	View Code Duplication	class GSMModemCollection:
		0 ignored issues – show Duplication introduced 2020-11-24 10:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
10		@staticmethod
11		def __init__():
12		pass
13
14		@staticmethod
15		def on_options(req, resp):
16		resp.status = falcon.HTTP_200
17
18		@staticmethod
19		def on_get(req, resp):
20		cnx = mysql.connector.connect(**config.myems_fdd_db)
21		cursor = cnx.cursor()
22
23		query = (" SELECT id, serial_port, baud_rate "
24		" FROM tbl_gsm_modems ")
25		cursor.execute(query)
26		rows = cursor.fetchall()
27		cursor.close()
28		cnx.disconnect()
29
30		result = list()
31		if rows is not None and len(rows) > 0:
32		for row in rows:
33		meta_result = {"id": row[0],
34		"serial_port": row[1],
35		"baud_rate": row[2]}
36		result.append(meta_result)
37
38		resp.body = json.dumps(result)
39
40		@staticmethod
41		def on_post(req, resp):
42		"""Handles POST requests"""
43		try:
44		raw_json = req.stream.read().decode('utf-8')
45		except Exception as ex:
46		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR', description=ex)
47
48		new_values = json.loads(raw_json, encoding='utf-8')
49
50		if 'serial_port' not in new_values['data'].keys() or \
51		not isinstance(new_values['data']['serial_port'], str) or \
52		len(str.strip(new_values['data']['serial_port'])) == 0:
53		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
54		description='API.INVALID_SERIAL_PORT')
55
56		serial_port = str.strip(new_values['data']['serial_port'])
57
58		if 'baud_rate' not in new_values['data'].keys() or \
59		not isinstance(new_values['data']['baud_rate'], int) or \
60		new_values['data']['baud_rate'] <= 0:
61		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
62		description='API.INVALID_BAUD_RATE')
63		baud_rate = float(new_values['data']['baud_rate'])
64
65		cnx = mysql.connector.connect(**config.myems_fdd_db)
66		cursor = cnx.cursor()
67
68		cursor.execute(" SELECT id "
69		" FROM tbl_gsm_modems "
70		" WHERE serial_port = %s ", (serial_port,))
71		if cursor.fetchone() is not None:
72		cursor.close()
73		cnx.disconnect()
74		raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
75		description='API.GSM_MODEM_SERIAL_PORT_IS_ALREADY_IN_USE')
76
77		add_value = (" INSERT INTO tbl_gsm_modems "
78		" (serial_port, baud_rate) "
79		" VALUES (%s, %s) ")
80		cursor.execute(add_value, (serial_port,
81		baud_rate))
82		new_id = cursor.lastrowid
83		cnx.commit()
84		cursor.close()
85		cnx.disconnect()
86
87		resp.status = falcon.HTTP_201
88		resp.location = '/gsmmodems/' + str(new_id)
89
90
91		class GSMModemItem:
92		@staticmethod
93		def __init__():
94		pass
95
96		@staticmethod
97		def on_options(req, resp, id_):
98		resp.status = falcon.HTTP_200
99
100		@staticmethod
101		def on_get(req, resp, id_):
102		if not id_.isdigit() or int(id_) <= 0:
103		raise falcon.HTTPError(falcon.HTTP_400, '400 Bad Request')
104
105		cnx = mysql.connector.connect(**config.myems_fdd_db)
106		cursor = cnx.cursor()
107
108		query = (" SELECT id, serial_port, baud_rate "
109		" FROM tbl_gsm_modems "
110		" WHERE id = %s ")
111		cursor.execute(query, (id_,))
112		row = cursor.fetchone()
113		cursor.close()
114		cnx.disconnect()
115		if row is None:
116		raise falcon.HTTPError(falcon.HTTP_404, 'API.NOT_FOUND')
117
118		result = {"id": row[0],
119		"serial_port": row[1],
120		"baud_rate": row[2]}
121		resp.body = json.dumps(result)
122
123	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2020-11-24 10:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
124		def on_delete(req, resp, id_):
125		if not id_.isdigit() or int(id_) <= 0:
126		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
127		description='API.INVALID_GSM_MODEM_ID')
128
129		cnx = mysql.connector.connect(**config.myems_fdd_db)
130		cursor = cnx.cursor()
131
132		cursor.execute(" SELECT serial_port "
133		" FROM tbl_gsm_modems "
134		" WHERE id = %s ", (id_,))
135		if cursor.fetchone() is None:
136		cursor.close()
137		cnx.disconnect()
138		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
139		description='API.GSM_MODEM_NOT_FOUND')
140
141		cursor.execute(" DELETE FROM tbl_gsm_modems WHERE id = %s ", (id_,))
142		cnx.commit()
143
144		cursor.close()
145		cnx.disconnect()
146
147		resp.status = falcon.HTTP_204
148
149	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2020-11-24 10:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
150		def on_put(req, resp, id_):
151		"""Handles PUT requests"""
152		try:
153		raw_json = req.stream.read().decode('utf-8')
154		except Exception as ex:
155		raise falcon.HTTPError(falcon.HTTP_400, title='API.EXCEPTION', description=ex)
156
157		if not id_.isdigit() or int(id_) <= 0:
158		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
159		description='API.INVALID_GSM_MODEM_ID')
160
161		new_values = json.loads(raw_json, encoding='utf-8')
162
163		if 'serial_port' not in new_values['data'].keys() or \
164		not isinstance(new_values['data']['serial_port'], str) or \
165		len(str.strip(new_values['data']['serial_port'])) == 0:
166		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
167		description='API.INVALID_SERIAL_PORT')
168
169		serial_port = str.strip(new_values['data']['serial_port'])
170
171		if 'baud_rate' not in new_values['data'].keys() or \
172		not isinstance(new_values['data']['baud_rate'], int) or \
173		new_values['data']['baud_rate'] <= 0:
174		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
175		description='API.INVALID_BAUD_RATE')
176		baud_rate = float(new_values['data']['baud_rate'])
177
178		cnx = mysql.connector.connect(**config.myems_fdd_db)
179		cursor = cnx.cursor()
180
181		cursor.execute(" SELECT serial_port "
182		" FROM tbl_gsm_modems "
183		" WHERE id = %s ",
184		(id_,))
185		if cursor.fetchone() is None:
186		cursor.close()
187		cnx.disconnect()
188		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
189		description='API.GSM_MODEM_NOT_FOUND')
190
191		cursor.execute(" SELECT serial_port "
192		" FROM tbl_gsm_modems "
193		" WHERE serial_port = %s AND id != %s ", (serial_port, id_))
194		if cursor.fetchone() is not None:
195		cursor.close()
196		cnx.disconnect()
197		raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
198		description='API.GSM_MODEM_SERIAL_PORT_IS_ALREADY_IN_USE')
199
200		update_row = (" UPDATE tbl_gsm_modems "
201		" SET serial_port = %s, baud_rate = %s "
202		" WHERE id = %s ")
203		cursor.execute(update_row, (serial_port,
204		baud_rate,
205		id_,))
206		cnx.commit()
207
208		cursor.close()
209		cnx.disconnect()
210
211		resp.status = falcon.HTTP_200
212

myems / myems-api

Push — master ( e07a89...15a904 )

gsmmodem.GSMModemCollection.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

gsmmodem.GSMModemCollection.init() A