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

Code Duplication Length = 77-87 lines in 2 locations

myems-api/core/protocol.py 1 location



        resp.text = json.dumps(result)

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

        Creates a new protocol with the specified name and code.
        Validates that both name and code are unique.

        Args:
            req: Falcon request object containing protocol data:
                - name: Protocol name (required)
                - code: Protocol code (required)
            resp: Falcon response object
        """
        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')

        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 name already exists
        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')

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

        # Insert new protocol
        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:

myems-api/core/privilege.py 1 location



        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:

		@@ 72-158 (lines=87) @@
69
70		resp.text = json.dumps(result)
71
72		@staticmethod
73		@user_logger
74		def on_post(req, resp):
75		"""
76		Handle POST requests to create a new protocol
77
78		Creates a new protocol with the specified name and code.
79		Validates that both name and code are unique.
80
81		Args:
82		req: Falcon request object containing protocol data:
83		- name: Protocol name (required)
84		- code: Protocol code (required)
85		resp: Falcon response object
86		"""
87		admin_control(req)
88
89		# Read and parse request body
90		try:
91		raw_json = req.stream.read().decode('utf-8')
92		except UnicodeDecodeError as ex:
93		print("Failed to decode request")
94		raise falcon.HTTPError(status=falcon.HTTP_400,
95		title='API.BAD_REQUEST',
96		description='API.INVALID_ENCODING')
97		except Exception as ex:
98		print("Unexpected error reading request stream")
99		raise falcon.HTTPError(status=falcon.HTTP_400,
100		title='API.BAD_REQUEST',
101		description='API.FAILED_TO_READ_REQUEST_STREAM')
102
103		new_values = json.loads(raw_json)
104
105		# Validate protocol name
106		if 'name' not in new_values['data'].keys() or \
107		not isinstance(new_values['data']['name'], str) or \
108		len(str.strip(new_values['data']['name'])) == 0:
109		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
110		description='API.INVALID_PROTOCOL_NAME')
111		name = str.strip(new_values['data']['name'])
112
113		# Validate protocol code
114		if 'code' not in new_values['data'].keys() or \
115		not isinstance(new_values['data']['code'], str) or \
116		len(str.strip(new_values['data']['code'])) == 0:
117		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
118		description='API.INVALID_PROTOCOL_CODE')
119		code = str.strip(new_values['data']['code'])
120
121		# Connect to database
122		cnx = mysql.connector.connect(**config.myems_system_db)
123		cursor = cnx.cursor()
124
125		# Check if protocol name already exists
126		cursor.execute(" SELECT name "
127		" FROM tbl_protocols "
128		" WHERE name = %s ", (name,))
129		if cursor.fetchone() is not None:
130		cursor.close()
131		cnx.close()
132		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
133		description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
134
135		# Check if protocol code already exists
136		cursor.execute(" SELECT code "
137		" FROM tbl_protocols "
138		" WHERE code = %s ", (code,))
139		if cursor.fetchone() is not None:
140		cursor.close()
141		cnx.close()
142		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
143		description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
144
145		# Insert new protocol
146		add_row = (" INSERT INTO tbl_protocols "
147		" (name, code) "
148		" VALUES (%s, %s) ")
149
150		cursor.execute(add_row, (name,
151		code))
152		new_id = cursor.lastrowid
153		cnx.commit()
154		cursor.close()
155		cnx.close()
156
157		resp.status = falcon.HTTP_201
158		resp.location = '/protocols/' + str(new_id)
159
160
161		class ProtocolItem:

		@@ 69-145 (lines=77) @@
66
67		resp.text = json.dumps(result)
68
69		@staticmethod
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:

MyEMS / myems

Code Duplication Length = 77-87 lines in 2 locations

myems-api/core/protocol.py 1 location

myems-api/core/privilege.py 1 location