Code Duplication - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 90-102 lines in 2 locations

myems-api/core/protocol.py 1 location





    @staticmethod
    @user_logger
    def on_put(req, resp, id_):
        """
        Handle PUT requests to update a specific protocol

        Updates the protocol with the specified ID.
        Validates that the new name and code are unique.

        Args:
            req: Falcon request object containing updated protocol data:
                - name: Updated protocol name (required)
                - code: Updated protocol code (required)
            resp: Falcon response object
            id_: Protocol ID to update
        """
        admin_control(req)

        # Read and parse request body
        try:
            raw_json = req.stream.read().decode('utf-8')
        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 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 protocol ID
        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)

        # Validate protocol name
        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'])

        # Validate protocol code
        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'])

        # Connect to database
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Check if protocol exists
        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')

        # Check if new name already exists (excluding current protocol)
        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')

        # Check if new code already exists (excluding current protocol)
        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 the protocol
        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




myems-api/core/privilege.py 1 location



        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



		@@ 292-393 (lines=102) @@
289
290
291
292		@staticmethod
293		@user_logger
294		def on_put(req, resp, id_):
295		"""
296		Handle PUT requests to update a specific protocol
297
298		Updates the protocol with the specified ID.
299		Validates that the new name and code are unique.
300
301		Args:
302		req: Falcon request object containing updated protocol data:
303		- name: Updated protocol name (required)
304		- code: Updated protocol code (required)
305		resp: Falcon response object
306		id_: Protocol ID to update
307		"""
308		admin_control(req)
309
310		# Read and parse request body
311		try:
312		raw_json = req.stream.read().decode('utf-8')
313		except UnicodeDecodeError as ex:
314		print("Failed to decode request")
315		raise falcon.HTTPError(status=falcon.HTTP_400,
316		title='API.BAD_REQUEST',
317		description='API.INVALID_ENCODING')
318		except Exception as ex:
319		print("Unexpected error reading request stream")
320		raise falcon.HTTPError(status=falcon.HTTP_400,
321		title='API.BAD_REQUEST',
322		description='API.FAILED_TO_READ_REQUEST_STREAM')
323
324		# Validate protocol ID
325		if not id_.isdigit() or int(id_) <= 0:
326		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
327		description='API.INVALID_PROTOCOL_ID')
328
329		new_values = json.loads(raw_json)
330
331		# Validate protocol name
332		if 'name' not in new_values['data'].keys() or \
333		not isinstance(new_values['data']['name'], str) or \
334		len(str.strip(new_values['data']['name'])) == 0:
335		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
336		description='API.INVALID_PROTOCOL_NAME')
337		name = str.strip(new_values['data']['name'])
338
339		# Validate protocol code
340		if 'code' not in new_values['data'].keys() or \
341		not isinstance(new_values['data']['code'], str) or \
342		len(str.strip(new_values['data']['code'])) == 0:
343		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
344		description='API.INVALID_PROTOCOL_CODE')
345		code = str.strip(new_values['data']['code'])
346
347		# Connect to database
348		cnx = mysql.connector.connect(**config.myems_system_db)
349		cursor = cnx.cursor()
350
351		# Check if protocol exists
352		cursor.execute(" SELECT name "
353		" FROM tbl_protocols "
354		" WHERE id = %s ", (id_,))
355		if cursor.fetchone() is None:
356		cursor.close()
357		cnx.close()
358		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
359		description='API.PROTOCOL_NOT_FOUND')
360
361		# Check if new name already exists (excluding current protocol)
362		cursor.execute(" SELECT name "
363		" FROM tbl_protocols "
364		" WHERE name = %s AND id != %s ", (name, id_))
365		if cursor.fetchone() is not None:
366		cursor.close()
367		cnx.close()
368		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
369		description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
370
371		# Check if new code already exists (excluding current protocol)
372		cursor.execute(" SELECT code "
373		" FROM tbl_protocols "
374		" WHERE code = %s AND id != %s ", (code, id_))
375		if cursor.fetchone() is not None:
376		cursor.close()
377		cnx.close()
378		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
379		description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
380
381		# Update the protocol
382		update_row = (" UPDATE tbl_protocols "
383		" SET name = %s, code = %s "
384		" WHERE id = %s ")
385		cursor.execute(update_row, (name,
386		code,
387		id_,))
388		cnx.commit()
389
390		cursor.close()
391		cnx.close()
392
393		resp.status = falcon.HTTP_200
394
395
396

		@@ 228-317 (lines=90) @@
225
226		resp.status = falcon.HTTP_204
227
228		@staticmethod
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

Code Duplication Length = 90-102 lines in 2 locations

myems-api/core/protocol.py 1 location

myems-api/core/privilege.py 1 location