core.iotsimcard.IoTSIMCardCollection.__init__() - Code Metrics - Inspection of "Merge branch 'develop'" - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 98975f...e9b166 )

unknown

created 2025-09-28 03:58 UTC

core.iotsimcard.IoTSIMCardCollection.init() A

↳ Parent: core.iotsimcard

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import re
import uuid
import falcon
import mysql.connector
import simplejson as json
from core.useractivity import user_logger, admin_control, access_control, api_key_control
import config


class IoTSIMCardCollection:
    def __init__(self):
        """"Initializes"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, iccid, imsi, operator, status, active_time, open_time, expiration_time, "
                 "        used_traffic, total_traffic, description "
                 " FROM tbl_iot_sim_cards "
                 " ORDER BY id ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0],
                               "iccid": row[1],
                               "imsi": row[2],
                               "operator": row[3],
                               "status": row[4],
                               "active_time": row[5],
                               "open_time": row[6],
                               "expiration_time": row[7],
                               "used_traffic": row[8],
                               "total_traffic": row[9],
                               "description": row[10]}
                result.append(meta_result)

        resp.text = json.dumps(result)

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

        new_values = json.loads(raw_json)

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

        if 'description' in new_values['data'].keys() and \
                new_values['data']['description'] is not None and \
                len(str(new_values['data']['description'])) > 0:
            description = str.strip(new_values['data']['description'])
        else:
            description = None

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

        cursor.execute(" SELECT iccid "
                       " FROM tbl_iot_sim_cards "
                       " WHERE iccid = %s ", (iccid,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.ICCID_IS_ALREADY_IN_USE')

        add_row = (" INSERT INTO tbl_iot_sim_cards "
                   "     (iccid, description) "
                   " VALUES (%s, %s) ")

        cursor.execute(add_row, (iccid, description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class IoTSIMCardItem:
    def __init__(self):
        """"Initializes"""
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_CONTACT_ID')

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

        query = (" SELECT id, iccid, imsi, operator, status, active_time, open_time, expiration_time, "
                 "        used_traffic, total_traffic, description "
                 " FROM tbl_iot_sim_cards "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.IOT_SIM_CARD_NOT_FOUND')

        result = {"id": row[0],
                  "iccid": row[1],
                  "imsi": row[2],
                  "operator": row[3],
                  "status": row[4],
                  "active_time": row[5],
                  "open_time": row[6],
                  "expiration_time": row[7],
                  "used_traffic": row[8],
                  "total_traffic": row[9],
                  "description": row[10]}
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_):
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IOT_SIM_CARD_ID')

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

        cursor.execute(" SELECT iccid "
                       " FROM tbl_iot_sim_cards "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.IOT_SIM_CARD_NOT_FOUND')

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

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204

    @staticmethod

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

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

        new_values = json.loads(raw_json)

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

        if 'description' in new_values['data'].keys() and \
                new_values['data']['description'] is not None and \
                len(str(new_values['data']['description'])) > 0:
            description = str.strip(new_values['data']['description'])
        else:
            description = None

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

        cursor.execute(" SELECT iccid "
                       " FROM tbl_iot_sim_cards "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.IOT_SIM_CARD_NOT_FOUND')

        cursor.execute(" SELECT iccid "
                       " FROM tbl_iot_sim_cards "
                       " WHERE iccid = %s AND id != %s ", (iccid, id_))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.ICCID_IS_ALREADY_IN_USE')

        update_row = (" UPDATE tbl_iot_sim_cards "
                      " SET iccid = %s, description = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (iccid,
                                    description,
                                    id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200



1		import re
2		import uuid
3		import falcon
4		import mysql.connector
5		import simplejson as json
6		from core.useractivity import user_logger, admin_control, access_control, api_key_control
7		import config
8
9
10		class IoTSIMCardCollection:
11		def __init__(self):
12		""""Initializes"""
13		pass
14
15		@staticmethod
16		def on_options(req, resp):
17		_ = req
18		resp.status = falcon.HTTP_200
19
20		@staticmethod
21		def on_get(req, resp):
22		if 'API-KEY' not in req.headers or \
23		not isinstance(req.headers['API-KEY'], str) or \
24		len(str.strip(req.headers['API-KEY'])) == 0:
25		access_control(req)
26		else:
27		api_key_control(req)
28		cnx = mysql.connector.connect(**config.myems_system_db)
29		cursor = cnx.cursor()
30
31		query = (" SELECT id, iccid, imsi, operator, status, active_time, open_time, expiration_time, "
32		" used_traffic, total_traffic, description "
33		" FROM tbl_iot_sim_cards "
34		" ORDER BY id ")
35		cursor.execute(query)
36		rows = cursor.fetchall()
37		cursor.close()
38		cnx.close()
39
40		result = list()
41		if rows is not None and len(rows) > 0:
42		for row in rows:
43		meta_result = {"id": row[0],
44		"iccid": row[1],
45		"imsi": row[2],
46		"operator": row[3],
47		"status": row[4],
48		"active_time": row[5],
49		"open_time": row[6],
50		"expiration_time": row[7],
51		"used_traffic": row[8],
52		"total_traffic": row[9],
53		"description": row[10]}
54		result.append(meta_result)
55
56		resp.text = json.dumps(result)
57
58		@staticmethod
59		@user_logger
60		def on_post(req, resp):
61		"""Handles POST requests"""
62		admin_control(req)
63		try:
64		raw_json = req.stream.read().decode('utf-8')
65		except Exception as ex:
66		print(ex)
67		raise falcon.HTTPError(status=falcon.HTTP_400,
68		title='API.BAD_REQUEST',
69		description='API.FAILED_TO_READ_REQUEST_STREAM')
70
71		new_values = json.loads(raw_json)
72
73		if 'iccid' not in new_values['data'].keys() or \
74		not isinstance(new_values['data']['iccid'], str) or \
75		len(str.strip(new_values['data']['iccid'])) == 0:
76		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
77		description='API.INVALID_ICCID')
78		iccid = str.strip(new_values['data']['iccid'])
79
80		if 'description' in new_values['data'].keys() and \
81		new_values['data']['description'] is not None and \
82		len(str(new_values['data']['description'])) > 0:
83		description = str.strip(new_values['data']['description'])
84		else:
85		description = None
86
87		cnx = mysql.connector.connect(**config.myems_system_db)
88		cursor = cnx.cursor()
89
90		cursor.execute(" SELECT iccid "
91		" FROM tbl_iot_sim_cards "
92		" WHERE iccid = %s ", (iccid,))
93		if cursor.fetchone() is not None:
94		cursor.close()
95		cnx.close()
96		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
97		description='API.ICCID_IS_ALREADY_IN_USE')
98
99		add_row = (" INSERT INTO tbl_iot_sim_cards "
100		" (iccid, description) "
101		" VALUES (%s, %s) ")
102
103		cursor.execute(add_row, (iccid, description))
104		new_id = cursor.lastrowid
105		cnx.commit()
106		cursor.close()
107		cnx.close()
108
109		resp.status = falcon.HTTP_201
110		resp.location = '/iotsimcards/' + str(new_id)
111
112
113		class IoTSIMCardItem:
114		def __init__(self):
115		""""Initializes"""
116		pass
117
118		@staticmethod
119		def on_options(req, resp, id_):
120		_ = id_
121		_ = req
122		resp.status = falcon.HTTP_200
123
124		@staticmethod
125		def on_get(req, resp, id_):
126		if 'API-KEY' not in req.headers or \
127		not isinstance(req.headers['API-KEY'], str) or \
128		len(str.strip(req.headers['API-KEY'])) == 0:
129		access_control(req)
130		else:
131		api_key_control(req)
132		if not id_.isdigit() or int(id_) <= 0:
133		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
134		description='API.INVALID_CONTACT_ID')
135
136		cnx = mysql.connector.connect(**config.myems_system_db)
137		cursor = cnx.cursor()
138
139		query = (" SELECT id, iccid, imsi, operator, status, active_time, open_time, expiration_time, "
140		" used_traffic, total_traffic, description "
141		" FROM tbl_iot_sim_cards "
142		" WHERE id = %s ")
143		cursor.execute(query, (id_,))
144		row = cursor.fetchone()
145		cursor.close()
146		cnx.close()
147
148		if row is None:
149		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
150		description='API.IOT_SIM_CARD_NOT_FOUND')
151
152		result = {"id": row[0],
153		"iccid": row[1],
154		"imsi": row[2],
155		"operator": row[3],
156		"status": row[4],
157		"active_time": row[5],
158		"open_time": row[6],
159		"expiration_time": row[7],
160		"used_traffic": row[8],
161		"total_traffic": row[9],
162		"description": row[10]}
163		resp.text = json.dumps(result)
164
165		@staticmethod
166		@user_logger
167		def on_delete(req, resp, id_):
168		admin_control(req)
169		if not id_.isdigit() or int(id_) <= 0:
170		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
171		description='API.INVALID_IOT_SIM_CARD_ID')
172
173		cnx = mysql.connector.connect(**config.myems_system_db)
174		cursor = cnx.cursor()
175
176		cursor.execute(" SELECT iccid "
177		" FROM tbl_iot_sim_cards "
178		" WHERE id = %s ", (id_,))
179		if cursor.fetchone() is None:
180		cursor.close()
181		cnx.close()
182		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
183		description='API.IOT_SIM_CARD_NOT_FOUND')
184
185		cursor.execute(" DELETE FROM tbl_iot_sim_cards WHERE id = %s ", (id_,))
186		cnx.commit()
187
188		cursor.close()
189		cnx.close()
190
191		resp.status = falcon.HTTP_204
192
193	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2024-06-17 15:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
194		@user_logger
195		def on_put(req, resp, id_):
196		"""Handles PUT requests"""
197		admin_control(req)
198		try:
199		raw_json = req.stream.read().decode('utf-8')
200		except Exception as ex:
201		print(ex)
202		raise falcon.HTTPError(status=falcon.HTTP_400,
203		title='API.BAD_REQUEST',
204		description='API.FAILED_TO_READ_REQUEST_STREAM')
205
206		if not id_.isdigit() or int(id_) <= 0:
207		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
208		description='API.INVALID_IOT_SIM_CARD_ID')
209
210		new_values = json.loads(raw_json)
211
212		if 'iccid' not in new_values['data'].keys() or \
213		not isinstance(new_values['data']['iccid'], str) or \
214		len(str.strip(new_values['data']['iccid'])) == 0:
215		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
216		description='API.INVALID_ICCID')
217		iccid = str.strip(new_values['data']['iccid'])
218
219		if 'description' in new_values['data'].keys() and \
220		new_values['data']['description'] is not None and \
221		len(str(new_values['data']['description'])) > 0:
222		description = str.strip(new_values['data']['description'])
223		else:
224		description = None
225
226		cnx = mysql.connector.connect(**config.myems_system_db)
227		cursor = cnx.cursor()
228
229		cursor.execute(" SELECT iccid "
230		" FROM tbl_iot_sim_cards "
231		" WHERE id = %s ", (id_,))
232		if cursor.fetchone() is None:
233		cursor.close()
234		cnx.close()
235		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
236		description='API.IOT_SIM_CARD_NOT_FOUND')
237
238		cursor.execute(" SELECT iccid "
239		" FROM tbl_iot_sim_cards "
240		" WHERE iccid = %s AND id != %s ", (iccid, id_))
241		if cursor.fetchone() is not None:
242		cursor.close()
243		cnx.close()
244		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
245		description='API.ICCID_IS_ALREADY_IN_USE')
246
247		update_row = (" UPDATE tbl_iot_sim_cards "
248		" SET iccid = %s, description = %s "
249		" WHERE id = %s ")
250		cursor.execute(update_row, (iccid,
251		description,
252		id_,))
253		cnx.commit()
254
255		cursor.close()
256		cnx.close()
257
258		resp.status = falcon.HTTP_200
259
260

MyEMS / myems

Push — master ( 98975f...e9b166 )

core.iotsimcard.IoTSIMCardCollection.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

core.iotsimcard.IoTSIMCardCollection.init() A