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

Code Duplication Length = 139-139 lines in 2 locations

myems-api/core/equipment.py 1 location


from core.useractivity import user_logger, access_control


class EquipmentCollection:
    @staticmethod
    def __init__():
        """Initializes EquipmentCollection"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor(dictionary=True)

        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row['id']] = {"id": row['id'],
                                               "name": row['name'],
                                               "uuid": row['uuid']}

        query = (" SELECT id, name, uuid, "
                 "        is_input_counted, is_output_counted, "
                 "        cost_center_id, description "
                 " FROM tbl_equipments "
                 " ORDER BY id ")
        cursor.execute(query)
        rows_equipments = cursor.fetchall()

        result = list()
        if rows_equipments is not None and len(rows_equipments) > 0:
            for row in rows_equipments:
                cost_center = cost_center_dict.get(row['cost_center_id'], None)
                meta_result = {"id": row['id'],
                               "name": row['name'],
                               "uuid": row['uuid'],
                               "is_input_counted": bool(row['is_input_counted']),
                               "is_output_counted": bool(row['is_output_counted']),
                               "cost_center": cost_center,
                               "description": row['description'],
                               "qrcode": 'equipment:' + row['uuid']}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

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

        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(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_EQUIPMENT_NAME')
        name = str.strip(new_values['data']['name'])

        if 'is_input_counted' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_input_counted'], bool):
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_INPUT_COUNTED_VALUE')
        is_input_counted = new_values['data']['is_input_counted']

        if 'is_output_counted' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_output_counted'], bool):
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
        is_output_counted = new_values['data']['is_output_counted']

        if 'cost_center_id' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['cost_center_id'], int) or \
                new_values['data']['cost_center_id'] <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COST_CENTER_ID')
        cost_center_id = new_values['data']['cost_center_id']

        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 name "
                       " FROM tbl_equipments "
                       " WHERE name = %s ", (name,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.EQUIPMENT_NAME_IS_ALREADY_IN_USE')

        if cost_center_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_cost_centers "
                           " WHERE id = %s ",
                           (new_values['data']['cost_center_id'],))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.COST_CENTER_NOT_FOUND')

        add_values = (" INSERT INTO tbl_equipments "
                      "    (name, uuid, is_input_counted, is_output_counted, "
                      "     cost_center_id, description) "
                      " VALUES (%s, %s, %s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    is_input_counted,
                                    is_output_counted,
                                    cost_center_id,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class EquipmentItem:

myems-api/core/combinedequipment.py 1 location


from core.useractivity import user_logger, access_control


class CombinedEquipmentCollection:
    @staticmethod
    def __init__():
        """ Initializes CombinedEquipmentCollection"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row['id']] = {"id": row[0],
                                               "name": row[1],
                                               "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        is_input_counted, is_output_counted, "
                 "        cost_center_id, description "
                 " FROM tbl_combined_equipments "
                 " ORDER BY id ")
        cursor.execute(query)
        rows_combined_equipments = cursor.fetchall()

        result = list()
        if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
            for row in rows_combined_equipments:
                cost_center = cost_center_dict.get(row[5], None)
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "is_input_counted": bool(row[3]),
                               "is_output_counted": bool(row[4]),
                               "cost_center": cost_center,
                               "description": row[6],
                               "qrcode": 'combinedequipment:' + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

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

        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(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COMBINED_EQUIPMENT_NAME')
        name = str.strip(new_values['data']['name'])

        if 'is_input_counted' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_input_counted'], bool):
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_INPUT_COUNTED_VALUE')
        is_input_counted = new_values['data']['is_input_counted']

        if 'is_output_counted' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_output_counted'], bool):
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
        is_output_counted = new_values['data']['is_output_counted']

        if 'cost_center_id' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['cost_center_id'], int) or \
                new_values['data']['cost_center_id'] <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COST_CENTER_ID')
        cost_center_id = new_values['data']['cost_center_id']

        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 name "
                       " FROM tbl_combined_equipments "
                       " WHERE name = %s ", (name,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
                                   description='API.COMBINED_EQUIPMENT_NAME_IS_ALREADY_IN_USE')

        if cost_center_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_cost_centers "
                           " WHERE id = %s ",
                           (new_values['data']['cost_center_id'],))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.COST_CENTER_NOT_FOUND')

        add_values = (" INSERT INTO tbl_combined_equipments "
                      "    (name, uuid, is_input_counted, is_output_counted, "
                      "     cost_center_id, description) "
                      " VALUES (%s, %s, %s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    is_input_counted,
                                    is_output_counted,
                                    cost_center_id,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class CombinedEquipmentItem:

		@@ 9-147 (lines=139) @@
6		from core.useractivity import user_logger, access_control
7
8
9		class EquipmentCollection:
10		@staticmethod
11		def __init__():
12		"""Initializes EquipmentCollection"""
13		pass
14
15		@staticmethod
16		def on_options(req, resp):
17		resp.status = falcon.HTTP_200
18
19		@staticmethod
20		def on_get(req, resp):
21		cnx = mysql.connector.connect(**config.myems_system_db)
22		cursor = cnx.cursor(dictionary=True)
23
24		query = (" SELECT id, name, uuid "
25		" FROM tbl_cost_centers ")
26		cursor.execute(query)
27		rows_cost_centers = cursor.fetchall()
28
29		cost_center_dict = dict()
30		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
31		for row in rows_cost_centers:
32		cost_center_dict[row['id']] = {"id": row['id'],
33		"name": row['name'],
34		"uuid": row['uuid']}
35
36		query = (" SELECT id, name, uuid, "
37		" is_input_counted, is_output_counted, "
38		" cost_center_id, description "
39		" FROM tbl_equipments "
40		" ORDER BY id ")
41		cursor.execute(query)
42		rows_equipments = cursor.fetchall()
43
44		result = list()
45		if rows_equipments is not None and len(rows_equipments) > 0:
46		for row in rows_equipments:
47		cost_center = cost_center_dict.get(row['cost_center_id'], None)
48		meta_result = {"id": row['id'],
49		"name": row['name'],
50		"uuid": row['uuid'],
51		"is_input_counted": bool(row['is_input_counted']),
52		"is_output_counted": bool(row['is_output_counted']),
53		"cost_center": cost_center,
54		"description": row['description'],
55		"qrcode": 'equipment:' + row['uuid']}
56		result.append(meta_result)
57
58		cursor.close()
59		cnx.close()
60		resp.text = json.dumps(result)
61
62		@staticmethod
63		@user_logger
64		def on_post(req, resp):
65		"""Handles POST requests"""
66		access_control(req)
67		try:
68		raw_json = req.stream.read().decode('utf-8')
69		except Exception as ex:
70		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR', description=ex)
71
72		new_values = json.loads(raw_json)
73
74		if 'name' not in new_values['data'].keys() or \
75		not isinstance(new_values['data']['name'], str) or \
76		len(str.strip(new_values['data']['name'])) == 0:
77		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
78		description='API.INVALID_EQUIPMENT_NAME')
79		name = str.strip(new_values['data']['name'])
80
81		if 'is_input_counted' not in new_values['data'].keys() or \
82		not isinstance(new_values['data']['is_input_counted'], bool):
83		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
84		description='API.INVALID_IS_INPUT_COUNTED_VALUE')
85		is_input_counted = new_values['data']['is_input_counted']
86
87		if 'is_output_counted' not in new_values['data'].keys() or \
88		not isinstance(new_values['data']['is_output_counted'], bool):
89		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
90		description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
91		is_output_counted = new_values['data']['is_output_counted']
92
93		if 'cost_center_id' not in new_values['data'].keys() or \
94		not isinstance(new_values['data']['cost_center_id'], int) or \
95		new_values['data']['cost_center_id'] <= 0:
96		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
97		description='API.INVALID_COST_CENTER_ID')
98		cost_center_id = new_values['data']['cost_center_id']
99
100		if 'description' in new_values['data'].keys() and \
101		new_values['data']['description'] is not None and \
102		len(str(new_values['data']['description'])) > 0:
103		description = str.strip(new_values['data']['description'])
104		else:
105		description = None
106
107		cnx = mysql.connector.connect(**config.myems_system_db)
108		cursor = cnx.cursor()
109
110		cursor.execute(" SELECT name "
111		" FROM tbl_equipments "
112		" WHERE name = %s ", (name,))
113		if cursor.fetchone() is not None:
114		cursor.close()
115		cnx.close()
116		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
117		description='API.EQUIPMENT_NAME_IS_ALREADY_IN_USE')
118
119		if cost_center_id is not None:
120		cursor.execute(" SELECT name "
121		" FROM tbl_cost_centers "
122		" WHERE id = %s ",
123		(new_values['data']['cost_center_id'],))
124		row = cursor.fetchone()
125		if row is None:
126		cursor.close()
127		cnx.close()
128		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
129		description='API.COST_CENTER_NOT_FOUND')
130
131		add_values = (" INSERT INTO tbl_equipments "
132		" (name, uuid, is_input_counted, is_output_counted, "
133		" cost_center_id, description) "
134		" VALUES (%s, %s, %s, %s, %s, %s) ")
135		cursor.execute(add_values, (name,
136		str(uuid.uuid4()),
137		is_input_counted,
138		is_output_counted,
139		cost_center_id,
140		description))
141		new_id = cursor.lastrowid
142		cnx.commit()
143		cursor.close()
144		cnx.close()
145
146		resp.status = falcon.HTTP_201
147		resp.location = '/equipments/' + str(new_id)
148
149
150		class EquipmentItem:

		@@ 9-147 (lines=139) @@
6		from core.useractivity import user_logger, access_control
7
8
9		class CombinedEquipmentCollection:
10		@staticmethod
11		def __init__():
12		""" Initializes CombinedEquipmentCollection"""
13		pass
14
15		@staticmethod
16		def on_options(req, resp):
17		resp.status = falcon.HTTP_200
18
19		@staticmethod
20		def on_get(req, resp):
21		cnx = mysql.connector.connect(**config.myems_system_db)
22		cursor = cnx.cursor()
23
24		query = (" SELECT id, name, uuid "
25		" FROM tbl_cost_centers ")
26		cursor.execute(query)
27		rows_cost_centers = cursor.fetchall()
28
29		cost_center_dict = dict()
30		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
31		for row in rows_cost_centers:
32		cost_center_dict[row['id']] = {"id": row[0],
33		"name": row[1],
34		"uuid": row[2]}
35
36		query = (" SELECT id, name, uuid, "
37		" is_input_counted, is_output_counted, "
38		" cost_center_id, description "
39		" FROM tbl_combined_equipments "
40		" ORDER BY id ")
41		cursor.execute(query)
42		rows_combined_equipments = cursor.fetchall()
43
44		result = list()
45		if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
46		for row in rows_combined_equipments:
47		cost_center = cost_center_dict.get(row[5], None)
48		meta_result = {"id": row[0],
49		"name": row[1],
50		"uuid": row[2],
51		"is_input_counted": bool(row[3]),
52		"is_output_counted": bool(row[4]),
53		"cost_center": cost_center,
54		"description": row[6],
55		"qrcode": 'combinedequipment:' + row[2]}
56		result.append(meta_result)
57
58		cursor.close()
59		cnx.close()
60		resp.text = json.dumps(result)
61
62		@staticmethod
63		@user_logger
64		def on_post(req, resp):
65		"""Handles POST requests"""
66		access_control(req)
67		try:
68		raw_json = req.stream.read().decode('utf-8')
69		except Exception as ex:
70		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR', description=ex)
71
72		new_values = json.loads(raw_json)
73
74		if 'name' not in new_values['data'].keys() or \
75		not isinstance(new_values['data']['name'], str) or \
76		len(str.strip(new_values['data']['name'])) == 0:
77		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
78		description='API.INVALID_COMBINED_EQUIPMENT_NAME')
79		name = str.strip(new_values['data']['name'])
80
81		if 'is_input_counted' not in new_values['data'].keys() or \
82		not isinstance(new_values['data']['is_input_counted'], bool):
83		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
84		description='API.INVALID_IS_INPUT_COUNTED_VALUE')
85		is_input_counted = new_values['data']['is_input_counted']
86
87		if 'is_output_counted' not in new_values['data'].keys() or \
88		not isinstance(new_values['data']['is_output_counted'], bool):
89		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
90		description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
91		is_output_counted = new_values['data']['is_output_counted']
92
93		if 'cost_center_id' not in new_values['data'].keys() or \
94		not isinstance(new_values['data']['cost_center_id'], int) or \
95		new_values['data']['cost_center_id'] <= 0:
96		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
97		description='API.INVALID_COST_CENTER_ID')
98		cost_center_id = new_values['data']['cost_center_id']
99
100		if 'description' in new_values['data'].keys() and \
101		new_values['data']['description'] is not None and \
102		len(str(new_values['data']['description'])) > 0:
103		description = str.strip(new_values['data']['description'])
104		else:
105		description = None
106
107		cnx = mysql.connector.connect(**config.myems_system_db)
108		cursor = cnx.cursor()
109
110		cursor.execute(" SELECT name "
111		" FROM tbl_combined_equipments "
112		" WHERE name = %s ", (name,))
113		if cursor.fetchone() is not None:
114		cursor.close()
115		cnx.close()
116		raise falcon.HTTPError(falcon.HTTP_404, title='API.BAD_REQUEST',
117		description='API.COMBINED_EQUIPMENT_NAME_IS_ALREADY_IN_USE')
118
119		if cost_center_id is not None:
120		cursor.execute(" SELECT name "
121		" FROM tbl_cost_centers "
122		" WHERE id = %s ",
123		(new_values['data']['cost_center_id'],))
124		row = cursor.fetchone()
125		if row is None:
126		cursor.close()
127		cnx.close()
128		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
129		description='API.COST_CENTER_NOT_FOUND')
130
131		add_values = (" INSERT INTO tbl_combined_equipments "
132		" (name, uuid, is_input_counted, is_output_counted, "
133		" cost_center_id, description) "
134		" VALUES (%s, %s, %s, %s, %s, %s) ")
135		cursor.execute(add_values, (name,
136		str(uuid.uuid4()),
137		is_input_counted,
138		is_output_counted,
139		cost_center_id,
140		description))
141		new_id = cursor.lastrowid
142		cnx.commit()
143		cursor.close()
144		cnx.close()
145
146		resp.status = falcon.HTTP_201
147		resp.location = '/combinedequipments/' + str(new_id)
148
149
150		class CombinedEquipmentItem:

MyEMS / myems

Code Duplication Length = 139-139 lines in 2 locations

myems-api/core/equipment.py 1 location

myems-api/core/combinedequipment.py 1 location