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

Code Duplication Length = 135-135 lines in 2 locations

combinedequipment.py 1 location


import uuid


class CombinedEquipmentCollection:
    @staticmethod
    def __init__():
        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_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['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']}
                result.append(meta_result)

        cursor.close()
        cnx.disconnect()
        resp.body = json.dumps(result)

    @staticmethod
    def on_post(req, resp):
        """Handles POST requests"""
        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, encoding='utf-8')

        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.disconnect()
            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.disconnect()
                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.disconnect()

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


class CombinedEquipmentItem:

equipment.py 1 location


import uuid


class EquipmentCollection:
    @staticmethod
    def __init__():
        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']}
                result.append(meta_result)

        cursor.close()
        cnx.disconnect()
        resp.body = json.dumps(result)

    @staticmethod
    def on_post(req, resp):
        """Handles POST requests"""
        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, encoding='utf-8')

        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.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, 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.disconnect()
                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.disconnect()

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


class EquipmentItem:

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

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

myems / myems-api

Code Duplication Length = 135-135 lines in 2 locations

combinedequipment.py 1 location

equipment.py 1 location