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

Passed

Push — master ( 42322a...183c79 )

unknown

created 2025-07-21 00:41 UTC

core.protocol.ProtocolItem.on_options() A

↳ Parent: core.protocol

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

from datetime import datetime, timedelta
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 ProtocolCollection:
    def __init__(self):
        """"Initializes ContactCollection"""
        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, name, code "
                 " FROM tbl_protocols "
                 " 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],
                               "name": row[1],
                               "code": row[2]}
                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(str(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 'name' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['name'], str) or \
                len(str.strip(new_values['data']['name'])) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PROTOCOL_NAME')
        name = str.strip(new_values['data']['name'])

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

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

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

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

        add_row = (" INSERT INTO tbl_protocols "
                   "     (name, code) "
                   " VALUES (%s, %s) ")

        cursor.execute(add_row, (name,
                                 code))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class ProtocolItem:
    def __init__(self):
        """"Initializes ContactItem"""
        pass

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

    @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_PROTOCOL_ID')

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

        query = (" SELECT id, name, code "
                 " FROM tbl_protocols "
                 " 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.PROTOCOL_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "code": row[2]}
        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_PROTOCOL_ID'
            )

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()
        cursor.execute("SELECT name,code FROM tbl_protocols WHERE id = %s", (id_,))
        row = cursor.fetchone()
        if row is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(
                status=falcon.HTTP_404,
                title='API.NOT_FOUND',
                description='API.PROTOCOL_NOT_FOUND'
            )

        # check if this protocol is being used by any data sources
        code=row[1]
        cursor.execute(" SELECT name "
                       " FROM tbl_data_sources "
                       " WHERE protocol = %s "
                       " LIMIT 1 ",
                       (code,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.THERE_IS_RELATION_WITH_DATA_SOURCES')
        cursor.execute(" DELETE FROM tbl_protocols 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(str(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_PROTOCOL_ID')

        new_values = json.loads(raw_json)

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

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

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

        cursor.execute(" SELECT name "
                       " FROM tbl_protocols "
                       " 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.PROTOCOL_NOT_FOUND')

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

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

        update_row = (" UPDATE tbl_protocols "
                      " SET name = %s, code = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (name,
                                    code,
                                    id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200





1			from datetime import datetime, timedelta
2			import falcon
3			import mysql.connector
4			import simplejson as json
5			from core.useractivity import user_logger, admin_control, access_control, api_key_control
6			import config
7
8
9			class ProtocolCollection:
10			def __init__(self):
11			""""Initializes ContactCollection"""
12			pass
13
14			@staticmethod
15			def on_options(req, resp):
16			_ = req
17			resp.status = falcon.HTTP_200
18
19			@staticmethod
20			def on_get(req, resp):
21			if 'API-KEY' not in req.headers or \
22			not isinstance(req.headers['API-KEY'], str) or \
23			len(str.strip(req.headers['API-KEY'])) == 0:
24			access_control(req)
25			else:
26			api_key_control(req)
27			cnx = mysql.connector.connect(**config.myems_system_db)
28			cursor = cnx.cursor()
29
30			query = (" SELECT id, name, code "
31			" FROM tbl_protocols "
32			" ORDER BY id ")
33			cursor.execute(query)
34			rows = cursor.fetchall()
35			cursor.close()
36			cnx.close()
37
38			result = list()
39			if rows is not None and len(rows) > 0:
40			for row in rows:
41			meta_result = {"id": row[0],
42			"name": row[1],
43			"code": row[2]}
44			result.append(meta_result)
45
46			resp.text = json.dumps(result)
47
48			@staticmethod
49			@user_logger
50			def on_post(req, resp):
51			"""Handles POST requests"""
52			admin_control(req)
53			try:
54			raw_json = req.stream.read().decode('utf-8')
55			except Exception as ex:
56			print(str(ex))
57			raise falcon.HTTPError(status=falcon.HTTP_400,
58			title='API.BAD_REQUEST',
59			description='API.FAILED_TO_READ_REQUEST_STREAM')
60
61			new_values = json.loads(raw_json)
62
63			if 'name' not in new_values['data'].keys() or \
64			not isinstance(new_values['data']['name'], str) or \
65			len(str.strip(new_values['data']['name'])) == 0:
66			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
67			description='API.INVALID_PROTOCOL_NAME')
68			name = str.strip(new_values['data']['name'])
69
70			if 'code' not in new_values['data'].keys() or \
71			not isinstance(new_values['data']['code'], str) or \
72			len(str.strip(new_values['data']['code'])) == 0:
73			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
74			description='API.INVALID_PROTOCOL_CODE')
75			code = str.strip(new_values['data']['code'])
76
77			cnx = mysql.connector.connect(**config.myems_system_db)
78			cursor = cnx.cursor()
79
80			cursor.execute(" SELECT name "
81			" FROM tbl_protocols "
82			" WHERE name = %s ", (name,))
83			if cursor.fetchone() is not None:
84			cursor.close()
85			cnx.close()
86			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
87			description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
88
89			cursor.execute(" SELECT code "
90			" FROM tbl_protocols "
91			" WHERE code = %s ", (code,))
92			if cursor.fetchone() is not None:
93			cursor.close()
94			cnx.close()
95			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
96			description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
97
98			add_row = (" INSERT INTO tbl_protocols "
99			" (name, code) "
100			" VALUES (%s, %s) ")
101
102			cursor.execute(add_row, (name,
103			code))
104			new_id = cursor.lastrowid
105			cnx.commit()
106			cursor.close()
107			cnx.close()
108
109			resp.status = falcon.HTTP_201
110			resp.location = '/protocols/' + str(new_id)
111
112
113			class ProtocolItem:
114			def __init__(self):
115			""""Initializes ContactItem"""
116			pass
117
118			@staticmethod
119			def on_options(req, resp, id_):
120			_ = req
121			resp.status = falcon.HTTP_200
122			_ = id_
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_PROTOCOL_ID')
135
136			cnx = mysql.connector.connect(**config.myems_system_db)
137			cursor = cnx.cursor()
138
139			query = (" SELECT id, name, code "
140			" FROM tbl_protocols "
141			" WHERE id = %s ")
142			cursor.execute(query, (id_,))
143			row = cursor.fetchone()
144			cursor.close()
145			cnx.close()
146
147			if row is None:
148			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
149			description='API.PROTOCOL_NOT_FOUND')
150
151			result = {"id": row[0],
152			"name": row[1],
153			"code": row[2]}
154			resp.text = json.dumps(result)
155
156			@staticmethod
157			@user_logger
158			def on_delete(req, resp, id_):
159			admin_control(req)
160			if not id_.isdigit() or int(id_) <= 0:
161			raise falcon.HTTPError(
162			status=falcon.HTTP_400,
163			title='API.BAD_REQUEST',
164			description='API.INVALID_PROTOCOL_ID'
165			)
166
167			cnx = mysql.connector.connect(**config.myems_system_db)
168			cursor = cnx.cursor()
169			cursor.execute("SELECT name,code FROM tbl_protocols WHERE id = %s", (id_,))
170			row = cursor.fetchone()
171			if row is None:
172			cursor.close()
173			cnx.close()
174			raise falcon.HTTPError(
175			status=falcon.HTTP_404,
176			title='API.NOT_FOUND',
177			description='API.PROTOCOL_NOT_FOUND'
178			)
179
180			# check if this protocol is being used by any data sources
181			code=row[1]
182			cursor.execute(" SELECT name "
183			" FROM tbl_data_sources "
184			" WHERE protocol = %s "
185			" LIMIT 1 ",
186			(code,))
187			if cursor.fetchone() is not None:
188			cursor.close()
189			cnx.close()
190			raise falcon.HTTPError(status=falcon.HTTP_400,
191			title='API.BAD_REQUEST',
192			description='API.THERE_IS_RELATION_WITH_DATA_SOURCES')
193			cursor.execute(" DELETE FROM tbl_protocols WHERE id = %s ", (id_,))
194			cnx.commit()
195
196			cursor.close()
197			cnx.close()
198			resp.status = falcon.HTTP_204
199
200
201
202			@staticmethod
203			@user_logger
204			def on_put(req, resp, id_):
205			"""Handles PUT requests"""
206			admin_control(req)
207			try:
208			raw_json = req.stream.read().decode('utf-8')
209			except Exception as ex:
210			print(str(ex))
211			raise falcon.HTTPError(status=falcon.HTTP_400,
212			title='API.BAD_REQUEST',
213			description='API.FAILED_TO_READ_REQUEST_STREAM')
214
215			if not id_.isdigit() or int(id_) <= 0:
216			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
217			description='API.INVALID_PROTOCOL_ID')
218
219			new_values = json.loads(raw_json)
220
221			if 'name' not in new_values['data'].keys() or \
222			not isinstance(new_values['data']['name'], str) or \
223			len(str.strip(new_values['data']['name'])) == 0:
224			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
225			description='API.INVALID_PROTOCOL_NAME')
226			name = str.strip(new_values['data']['name'])
227
228			if 'code' not in new_values['data'].keys() or \
229			not isinstance(new_values['data']['code'], str) or \
230			len(str.strip(new_values['data']['code'])) == 0:
231			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
232			description='API.INVALID_PROTOCOL_CODE')
233			code = str.strip(new_values['data']['code'])
234
235			cnx = mysql.connector.connect(**config.myems_system_db)
236			cursor = cnx.cursor()
237
238			cursor.execute(" SELECT name "
239			" FROM tbl_protocols "
240			" WHERE id = %s ", (id_,))
241			if cursor.fetchone() is None:
242			cursor.close()
243			cnx.close()
244			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
245			description='API.PROTOCOL_NOT_FOUND')
246
247			cursor.execute(" SELECT name "
248			" FROM tbl_protocols "
249			" WHERE name = %s AND id != %s ", (name, id_))
250			if cursor.fetchone() is not None:
251			cursor.close()
252			cnx.close()
253			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
254			description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
255
256			cursor.execute(" SELECT code "
257			" FROM tbl_protocols "
258			" WHERE code = %s AND id != %s ", (code, id_))
259			if cursor.fetchone() is not None:
260			cursor.close()
261			cnx.close()
262			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
263			description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
264
265			update_row = (" UPDATE tbl_protocols "
266			" SET name = %s, code = %s "
267			" WHERE id = %s ")
268			cursor.execute(update_row, (name,
269			code,
270			id_,))
271			cnx.commit()
272
273			cursor.close()
274			cnx.close()
275
276			resp.status = falcon.HTTP_200
277
278
279
280

MyEMS / myems

Push — master ( 42322a...183c79 )

core.protocol.ProtocolItem.on_options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like