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

Passed

Push — master ( cb176a...4b8481 )

unknown

created 2025-12-17 14:15 UTC

core.privilege.PrivilegeCollection.on_post() C

↳ Parent: core.privilege

Complexity

Conditions

Size

Total Lines	77
Code Lines	51

Duplication

Lines	77
Ratio	100 %

Importance

Changes

Metric	Value
eloc	51
dl	77
loc	77
rs	5.3836
c	0
b	0
f	0
cc	11
nop	2

How to fix Long Method Complexity

import falcon
import mysql.connector
import simplejson as json
from core.useractivity import user_logger, admin_control
import config


class PrivilegeCollection:
    """
    Privilege Collection Resource

    This class handles privilege management operations for the MyEMS system.
    It provides functionality to create and retrieve user privileges
    that define access permissions and roles.
    """

    def __init__(self):
        pass

    @staticmethod
    def on_options(req, resp):
        """
        Handle OPTIONS request for CORS preflight

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all privileges

        Returns a list of all privileges with their metadata including:
        - Privilege ID and name
        - Privilege data (JSON configuration)

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        admin_control(req)
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor()

        # Query to retrieve all privileges ordered by ID descending
        query = (" SELECT id, name, data "
                 " FROM tbl_privileges "
                 " ORDER BY id DESC ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

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

        resp.text = json.dumps(result)

    @staticmethod

    @user_logger
    def on_post(req, resp):
        """
        Handle POST requests to create a new privilege

        Creates a new privilege with the specified name and data configuration.
        Requires admin privileges.

        Args:
            req: Falcon request object containing privilege data:
                - name: Privilege name (required)
                - data: Privilege data configuration (required)
            resp: Falcon response object
        """
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
            new_values = json.loads(raw_json)
        except UnicodeDecodeError as ex:
            print("Failed to decode request")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_ENCODING')
        except json.JSONDecodeError as ex:
            print("Failed to parse JSON")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_JSON_FORMAT')
        except Exception as ex:
            print("Unexpected error reading request stream")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')

        # Validate privilege name
        if 'name' not in new_values['data'] 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_PRIVILEGE_NAME')
        name = str.strip(new_values['data']['name'])

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

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

        # Check if privilege name already exists
        cursor.execute(" SELECT name "
                       " FROM tbl_privileges "
                       " 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.PRIVILEGE_NAME_IS_ALREADY_IN_USE')

        # Insert new privilege into database
        add_row = (" INSERT INTO tbl_privileges "
                   "             (name, data) "
                   " VALUES (%s, %s) ")

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

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


class PrivilegeItem:
    """
    Privilege Item Resource

    This class handles individual privilege operations including:
    - Updating privilege information
    - Deleting privileges (with relationship checks)
    """

    def __init__(self):
        pass

    @staticmethod
    def on_options(req, resp, id_):
        """
        Handle OPTIONS request for CORS preflight

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: Privilege ID parameter
        """
        _ = req
        resp.status = falcon.HTTP_200
        _ = id_

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_):
        """
        Handle DELETE requests to remove a privilege

        Deletes the specified privilege from the database.
        Checks for existing relationships with users before deletion.
        Requires admin privileges.

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: Privilege ID to delete
        """
        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_PRIVILEGE_ID')

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

        # Check for relationships with users
        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE privilege_id = %s ", (id_,))
        rows_users = cursor.fetchall()
        if rows_users is not None and len(rows_users) > 0:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.THERE_IS_RELATION_WITH_USERS')

        # Check if privilege exists
        cursor.execute(" SELECT name "
                       " FROM tbl_privileges "
                       " 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.PRIVILEGE_NOT_FOUND')

        # TODO: Delete associated objects before deleting privilege
        cursor.execute(" DELETE FROM tbl_privileges WHERE id = %s ", (id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204

    @staticmethod

    @user_logger
    def on_put(req, resp, id_):
        """
        Handle PUT requests to update privilege information

        Updates an existing privilege with new name and data configuration.
        Requires admin privileges.

        Args:
            req: Falcon request object containing update data:
                - name: New privilege name (required)
                - data: New privilege data configuration (required)
            resp: Falcon response object
            id_: Privilege ID to update
        """
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
            new_values = json.loads(raw_json)
        except UnicodeDecodeError as ex:
            print("Failed to decode request")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_ENCODING')
        except json.JSONDecodeError as ex:
            print("Failed to parse JSON")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_JSON_FORMAT')
        except Exception as ex:
            print("Unexpected error reading request stream")
            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_PRIVILEGE_ID')

        # Validate privilege name
        if 'name' not in new_values['data'] 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_PRIVILEGE_NAME')
        name = str.strip(new_values['data']['name'])

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

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

        # Check if privilege exists
        cursor.execute(" SELECT name "
                       " FROM tbl_privileges "
                       " 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.PRIVILEGE_NOT_FOUND')

        # Check if new name conflicts with existing privileges (excluding current)
        cursor.execute(" SELECT name "
                       " FROM tbl_privileges "
                       " 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.PRIVILEGE_NAME_IS_ALREADY_IN_USE')

        # Update privilege information
        update_row = (" UPDATE tbl_privileges "
                      " SET name = %s, data = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (name, data, id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200



1		import falcon
2		import mysql.connector
3		import simplejson as json
4		from core.useractivity import user_logger, admin_control
5		import config
6
7
8		class PrivilegeCollection:
9		"""
10		Privilege Collection Resource
11
12		This class handles privilege management operations for the MyEMS system.
13		It provides functionality to create and retrieve user privileges
14		that define access permissions and roles.
15		"""
16
17		def __init__(self):
18		pass
19
20		@staticmethod
21		def on_options(req, resp):
22		"""
23		Handle OPTIONS request for CORS preflight
24
25		Args:
26		req: Falcon request object
27		resp: Falcon response object
28		"""
29		_ = req
30		resp.status = falcon.HTTP_200
31
32		@staticmethod
33		def on_get(req, resp):
34		"""
35		Handle GET requests to retrieve all privileges
36
37		Returns a list of all privileges with their metadata including:
38		- Privilege ID and name
39		- Privilege data (JSON configuration)
40
41		Args:
42		req: Falcon request object
43		resp: Falcon response object
44		"""
45		admin_control(req)
46		cnx = mysql.connector.connect(**config.myems_user_db)
47		cursor = cnx.cursor()
48
49		# Query to retrieve all privileges ordered by ID descending
50		query = (" SELECT id, name, data "
51		" FROM tbl_privileges "
52		" ORDER BY id DESC ")
53		cursor.execute(query)
54		rows = cursor.fetchall()
55		cursor.close()
56		cnx.close()
57
58		# Build result list
59		result = list()
60		if rows is not None and len(rows) > 0:
61		for row in rows:
62		meta_result = {"id": row[0],
63		"name": row[1],
64		"data": row[2]}
65		result.append(meta_result)
66
67		resp.text = json.dumps(result)
68
69	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
70		@user_logger
71		def on_post(req, resp):
72		"""
73		Handle POST requests to create a new privilege
74
75		Creates a new privilege with the specified name and data configuration.
76		Requires admin privileges.
77
78		Args:
79		req: Falcon request object containing privilege data:
80		- name: Privilege name (required)
81		- data: Privilege data configuration (required)
82		resp: Falcon response object
83		"""
84		admin_control(req)
85		try:
86		raw_json = req.stream.read().decode('utf-8')
87		new_values = json.loads(raw_json)
88		except UnicodeDecodeError as ex:
89		print("Failed to decode request")
90		raise falcon.HTTPError(status=falcon.HTTP_400,
91		title='API.BAD_REQUEST',
92		description='API.INVALID_ENCODING')
93		except json.JSONDecodeError as ex:
94		print("Failed to parse JSON")
95		raise falcon.HTTPError(status=falcon.HTTP_400,
96		title='API.BAD_REQUEST',
97		description='API.INVALID_JSON_FORMAT')
98		except Exception as ex:
99		print("Unexpected error reading request stream")
100		raise falcon.HTTPError(status=falcon.HTTP_400,
101		title='API.BAD_REQUEST',
102		description='API.FAILED_TO_READ_REQUEST_STREAM')
103
104		# Validate privilege name
105		if 'name' not in new_values['data'] or \
106		not isinstance(new_values['data']['name'], str) or \
107		len(str.strip(new_values['data']['name'])) == 0:
108		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
109		description='API.INVALID_PRIVILEGE_NAME')
110		name = str.strip(new_values['data']['name'])
111
112		# Validate privilege data
113		if 'data' not in new_values['data'] or \
114		not isinstance(new_values['data']['data'], str) or \
115		len(str.strip(new_values['data']['data'])) == 0:
116		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
117		description='API.INVALID_PRIVILEGE_DATA')
118		data = str.strip(new_values['data']['data'])
119
120		cnx = mysql.connector.connect(**config.myems_user_db)
121		cursor = cnx.cursor()
122
123		# Check if privilege name already exists
124		cursor.execute(" SELECT name "
125		" FROM tbl_privileges "
126		" WHERE name = %s ", (name,))
127		if cursor.fetchone() is not None:
128		cursor.close()
129		cnx.close()
130		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
131		description='API.PRIVILEGE_NAME_IS_ALREADY_IN_USE')
132
133		# Insert new privilege into database
134		add_row = (" INSERT INTO tbl_privileges "
135		" (name, data) "
136		" VALUES (%s, %s) ")
137
138		cursor.execute(add_row, (name, data, ))
139		new_id = cursor.lastrowid
140		cnx.commit()
141		cursor.close()
142		cnx.close()
143
144		resp.status = falcon.HTTP_201
145		resp.location = '/privileges/' + str(new_id)
146
147
148		class PrivilegeItem:
149		"""
150		Privilege Item Resource
151
152		This class handles individual privilege operations including:
153		- Updating privilege information
154		- Deleting privileges (with relationship checks)
155		"""
156
157		def __init__(self):
158		pass
159
160		@staticmethod
161		def on_options(req, resp, id_):
162		"""
163		Handle OPTIONS request for CORS preflight
164
165		Args:
166		req: Falcon request object
167		resp: Falcon response object
168		id_: Privilege ID parameter
169		"""
170		_ = req
171		resp.status = falcon.HTTP_200
172		_ = id_
173
174		@staticmethod
175		@user_logger
176		def on_delete(req, resp, id_):
177		"""
178		Handle DELETE requests to remove a privilege
179
180		Deletes the specified privilege from the database.
181		Checks for existing relationships with users before deletion.
182		Requires admin privileges.
183
184		Args:
185		req: Falcon request object
186		resp: Falcon response object
187		id_: Privilege ID to delete
188		"""
189		admin_control(req)
190		if not id_.isdigit() or int(id_) <= 0:
191		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
192		description='API.INVALID_PRIVILEGE_ID')
193
194		cnx = mysql.connector.connect(**config.myems_user_db)
195		cursor = cnx.cursor()
196
197		# Check for relationships with users
198		cursor.execute(" SELECT id "
199		" FROM tbl_users "
200		" WHERE privilege_id = %s ", (id_,))
201		rows_users = cursor.fetchall()
202		if rows_users is not None and len(rows_users) > 0:
203		cursor.close()
204		cnx.close()
205		raise falcon.HTTPError(status=falcon.HTTP_400,
206		title='API.BAD_REQUEST',
207		description='API.THERE_IS_RELATION_WITH_USERS')
208
209		# Check if privilege exists
210		cursor.execute(" SELECT name "
211		" FROM tbl_privileges "
212		" WHERE id = %s ", (id_,))
213		if cursor.fetchone() is None:
214		cursor.close()
215		cnx.close()
216		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
217		description='API.PRIVILEGE_NOT_FOUND')
218
219		# TODO: Delete associated objects before deleting privilege
220		cursor.execute(" DELETE FROM tbl_privileges WHERE id = %s ", (id_,))
221		cnx.commit()
222
223		cursor.close()
224		cnx.close()
225
226		resp.status = falcon.HTTP_204
227
228	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
229		@user_logger
230		def on_put(req, resp, id_):
231		"""
232		Handle PUT requests to update privilege information
233
234		Updates an existing privilege with new name and data configuration.
235		Requires admin privileges.
236
237		Args:
238		req: Falcon request object containing update data:
239		- name: New privilege name (required)
240		- data: New privilege data configuration (required)
241		resp: Falcon response object
242		id_: Privilege ID to update
243		"""
244		admin_control(req)
245		try:
246		raw_json = req.stream.read().decode('utf-8')
247		new_values = json.loads(raw_json)
248		except UnicodeDecodeError as ex:
249		print("Failed to decode request")
250		raise falcon.HTTPError(status=falcon.HTTP_400,
251		title='API.BAD_REQUEST',
252		description='API.INVALID_ENCODING')
253		except json.JSONDecodeError as ex:
254		print("Failed to parse JSON")
255		raise falcon.HTTPError(status=falcon.HTTP_400,
256		title='API.BAD_REQUEST',
257		description='API.INVALID_JSON_FORMAT')
258		except Exception as ex:
259		print("Unexpected error reading request stream")
260		raise falcon.HTTPError(status=falcon.HTTP_400,
261		title='API.BAD_REQUEST',
262		description='API.FAILED_TO_READ_REQUEST_STREAM')
263
264		if not id_.isdigit() or int(id_) <= 0:
265		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
266		description='API.INVALID_PRIVILEGE_ID')
267
268		# Validate privilege name
269		if 'name' not in new_values['data'] or \
270		not isinstance(new_values['data']['name'], str) or \
271		len(str.strip(new_values['data']['name'])) == 0:
272		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
273		description='API.INVALID_PRIVILEGE_NAME')
274		name = str.strip(new_values['data']['name'])
275
276		# Validate privilege data
277		if 'data' not in new_values['data'] or \
278		not isinstance(new_values['data']['data'], str) or \
279		len(str.strip(new_values['data']['data'])) == 0:
280		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
281		description='API.INVALID_PRIVILEGE_DATA')
282		data = str.strip(new_values['data']['data'])
283
284		cnx = mysql.connector.connect(**config.myems_user_db)
285		cursor = cnx.cursor()
286
287		# Check if privilege exists
288		cursor.execute(" SELECT name "
289		" FROM tbl_privileges "
290		" WHERE id = %s ", (id_,))
291		if cursor.fetchone() is None:
292		cursor.close()
293		cnx.close()
294		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
295		description='API.PRIVILEGE_NOT_FOUND')
296
297		# Check if new name conflicts with existing privileges (excluding current)
298		cursor.execute(" SELECT name "
299		" FROM tbl_privileges "
300		" WHERE name = %s AND id != %s ", (name, id_))
301		if cursor.fetchone() is not None:
302		cursor.close()
303		cnx.close()
304		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
305		description='API.PRIVILEGE_NAME_IS_ALREADY_IN_USE')
306
307		# Update privilege information
308		update_row = (" UPDATE tbl_privileges "
309		" SET name = %s, data = %s "
310		" WHERE id = %s ")
311		cursor.execute(update_row, (name, data, id_,))
312		cnx.commit()
313
314		cursor.close()
315		cnx.close()
316
317		resp.status = falcon.HTTP_200
318
319

MyEMS / myems

Push — master ( cb176a...4b8481 )

core.privilege.PrivilegeCollection.on_post() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like