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

Code Duplication Length = 77-77 lines in 2 locations

myems-api/core/offlinemeter.py 2 locations


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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_OFFLINE_METER_ID')

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

        query = (" SELECT id, name, uuid "
                 " FROM tbl_energy_categories ")
        cursor.execute(query)
        rows_energy_categories = cursor.fetchall()

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

        query = (" SELECT id, name, uuid "
                 " FROM tbl_energy_items ")
        cursor.execute(query)
        rows_energy_items = cursor.fetchall()

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

        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[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        query = (" SELECT id, name, uuid, energy_category_id, "
                 "        is_counted, hourly_low_limit, hourly_high_limit, "
                 "        energy_item_id, cost_center_id, description "
                 " FROM tbl_offline_meters "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.OFFLINE_METER_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "energy_category": energy_category_dict.get(row[3], None),
                           "is_counted": True if row[4] else False,
                           "hourly_low_limit": row[5],
                           "hourly_high_limit": row[6],
                           "energy_item": energy_item_dict.get(row[7], None),
                           "cost_center": cost_center_dict.get(row[8], None),
                           "description": row[9]}

        resp.text = json.dumps(meta_result)


class OfflineMeterImport:

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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_OFFLINE_METER_ID')

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

        query = (" SELECT id, name, uuid "
                 " FROM tbl_energy_categories ")
        cursor.execute(query)
        rows_energy_categories = cursor.fetchall()

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

        query = (" SELECT id, name, uuid "
                 " FROM tbl_energy_items ")
        cursor.execute(query)
        rows_energy_items = cursor.fetchall()

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

        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[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        query = (" SELECT id, name, uuid, energy_category_id, "
                 "        is_counted, hourly_low_limit, hourly_high_limit, "
                 "        energy_item_id, cost_center_id, description "
                 " FROM tbl_offline_meters "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.OFFLINE_METER_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "energy_category": energy_category_dict.get(row[3], None),
                           "is_counted": True if row[4] else False,
                           "hourly_low_limit": row[5],
                           "hourly_high_limit": row[6],
                           "energy_item": energy_item_dict.get(row[7], None),
                           "cost_center": cost_center_dict.get(row[8], None),
                           "description": row[9]}

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger

		@@ 649-725 (lines=77) @@
646		def on_options(req, resp, id_):
647		resp.status = falcon.HTTP_200
648
649		@staticmethod
650		def on_get(req, resp, id_):
651		if 'API-KEY' not in req.headers or \
652		not isinstance(req.headers['API-KEY'], str) or \
653		len(str.strip(req.headers['API-KEY'])) == 0:
654		access_control(req)
655		else:
656		api_key_control(req)
657		if not id_.isdigit() or int(id_) <= 0:
658		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
659		description='API.INVALID_OFFLINE_METER_ID')
660
661		cnx = mysql.connector.connect(**config.myems_system_db)
662		cursor = cnx.cursor()
663
664		query = (" SELECT id, name, uuid "
665		" FROM tbl_energy_categories ")
666		cursor.execute(query)
667		rows_energy_categories = cursor.fetchall()
668
669		energy_category_dict = dict()
670		if rows_energy_categories is not None and len(rows_energy_categories) > 0:
671		for row in rows_energy_categories:
672		energy_category_dict[row[0]] = {"id": row[0],
673		"name": row[1],
674		"uuid": row[2]}
675
676		query = (" SELECT id, name, uuid "
677		" FROM tbl_energy_items ")
678		cursor.execute(query)
679		rows_energy_items = cursor.fetchall()
680
681		energy_item_dict = dict()
682		if rows_energy_items is not None and len(rows_energy_items) > 0:
683		for row in rows_energy_items:
684		energy_item_dict[row[0]] = {"id": row[0],
685		"name": row[1],
686		"uuid": row[2]}
687
688		query = (" SELECT id, name, uuid "
689		" FROM tbl_cost_centers ")
690		cursor.execute(query)
691		rows_cost_centers = cursor.fetchall()
692
693		cost_center_dict = dict()
694		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
695		for row in rows_cost_centers:
696		cost_center_dict[row[0]] = {"id": row[0],
697		"name": row[1],
698		"uuid": row[2]}
699
700		query = (" SELECT id, name, uuid, energy_category_id, "
701		" is_counted, hourly_low_limit, hourly_high_limit, "
702		" energy_item_id, cost_center_id, description "
703		" FROM tbl_offline_meters "
704		" WHERE id = %s ")
705		cursor.execute(query, (id_,))
706		row = cursor.fetchone()
707		cursor.close()
708		cnx.close()
709
710		if row is None:
711		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
712		description='API.OFFLINE_METER_NOT_FOUND')
713		else:
714		meta_result = {"id": row[0],
715		"name": row[1],
716		"uuid": row[2],
717		"energy_category": energy_category_dict.get(row[3], None),
718		"is_counted": True if row[4] else False,
719		"hourly_low_limit": row[5],
720		"hourly_high_limit": row[6],
721		"energy_item": energy_item_dict.get(row[7], None),
722		"cost_center": cost_center_dict.get(row[8], None),
723		"description": row[9]}
724
725		resp.text = json.dumps(meta_result)
726
727
728		class OfflineMeterImport:
		@@ 248-324 (lines=77) @@
245		def on_options(req, resp, id_):
246		resp.status = falcon.HTTP_200
247
248		@staticmethod
249		def on_get(req, resp, id_):
250		if 'API-KEY' not in req.headers or \
251		not isinstance(req.headers['API-KEY'], str) or \
252		len(str.strip(req.headers['API-KEY'])) == 0:
253		access_control(req)
254		else:
255		api_key_control(req)
256		if not id_.isdigit() or int(id_) <= 0:
257		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
258		description='API.INVALID_OFFLINE_METER_ID')
259
260		cnx = mysql.connector.connect(**config.myems_system_db)
261		cursor = cnx.cursor()
262
263		query = (" SELECT id, name, uuid "
264		" FROM tbl_energy_categories ")
265		cursor.execute(query)
266		rows_energy_categories = cursor.fetchall()
267
268		energy_category_dict = dict()
269		if rows_energy_categories is not None and len(rows_energy_categories) > 0:
270		for row in rows_energy_categories:
271		energy_category_dict[row[0]] = {"id": row[0],
272		"name": row[1],
273		"uuid": row[2]}
274
275		query = (" SELECT id, name, uuid "
276		" FROM tbl_energy_items ")
277		cursor.execute(query)
278		rows_energy_items = cursor.fetchall()
279
280		energy_item_dict = dict()
281		if rows_energy_items is not None and len(rows_energy_items) > 0:
282		for row in rows_energy_items:
283		energy_item_dict[row[0]] = {"id": row[0],
284		"name": row[1],
285		"uuid": row[2]}
286
287		query = (" SELECT id, name, uuid "
288		" FROM tbl_cost_centers ")
289		cursor.execute(query)
290		rows_cost_centers = cursor.fetchall()
291
292		cost_center_dict = dict()
293		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
294		for row in rows_cost_centers:
295		cost_center_dict[row[0]] = {"id": row[0],
296		"name": row[1],
297		"uuid": row[2]}
298
299		query = (" SELECT id, name, uuid, energy_category_id, "
300		" is_counted, hourly_low_limit, hourly_high_limit, "
301		" energy_item_id, cost_center_id, description "
302		" FROM tbl_offline_meters "
303		" WHERE id = %s ")
304		cursor.execute(query, (id_,))
305		row = cursor.fetchone()
306		cursor.close()
307		cnx.close()
308
309		if row is None:
310		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
311		description='API.OFFLINE_METER_NOT_FOUND')
312		else:
313		meta_result = {"id": row[0],
314		"name": row[1],
315		"uuid": row[2],
316		"energy_category": energy_category_dict.get(row[3], None),
317		"is_counted": True if row[4] else False,
318		"hourly_low_limit": row[5],
319		"hourly_high_limit": row[6],
320		"energy_item": energy_item_dict.get(row[7], None),
321		"cost_center": cost_center_dict.get(row[8], None),
322		"description": row[9]}
323
324		resp.text = json.dumps(meta_result)
325
326		@staticmethod
327		@user_logger

MyEMS / myems

Code Duplication Length = 77-77 lines in 2 locations

myems-api/core/offlinemeter.py 2 locations