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

Code Duplication Length = 111-115 lines in 2 locations

myems-api/core/energystoragepowerstation.py 1 location


        resp.status = falcon.HTTP_200


class EnergyStoragePowerStationContainerCollection:
    def __init__(self):
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')

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

        cursor.execute(" SELECT name "
                       " FROM tbl_energy_storage_power_stations "
                       " 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.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')

        query = (" SELECT s.id, s.name, s.uuid "
                 " FROM tbl_energy_storage_power_stations e, "
                 "      tbl_energy_storage_power_stations_containers es, tbl_energy_storage_containers s "
                 " WHERE es.energy_storage_power_station_id = e.id "
                 "       AND s.id = es.energy_storage_container_id "
                 "       AND e.id = %s "
                 " ORDER BY s.id ")
        cursor.execute(query, (id_,))
        rows = cursor.fetchall()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp, id_):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            print(str(ex))
            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_ENERGY_STORAGE_POWER_STATION_ID')

        new_values = json.loads(raw_json)

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

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

        cursor.execute(" SELECT name "
                       " FROM tbl_energy_storage_power_stations "
                       " 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.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')

        cursor.execute(" SELECT name "
                       " FROM tbl_energy_storage_containers "
                       " WHERE id = %s ", (energy_storage_container_id,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')

        query = (" SELECT id "
                 " FROM tbl_energy_storage_power_stations_containers "
                 " WHERE energy_storage_power_station_id = %s AND energy_storage_container_id = %s")
        cursor.execute(query, (id_, energy_storage_container_id,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.ENERGY_STORAGE_CONTAINER_SENSOR_RELATION_EXISTS')

        add_row = (" INSERT INTO tbl_energy_storage_power_stations_containers "
                   "        (energy_storage_power_station_id, energy_storage_container_id) "
                   " VALUES (%s, %s) ")
        cursor.execute(add_row, (id_, energy_storage_container_id,))
        cnx.commit()
        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_201
        resp.location = '/energystoragepowerstationss/' + str(id_) + '/containers/' + str(energy_storage_container_id)


class EnergyStoragePowerStationContainerItem:
    def __init__(self):

myems-api/core/microgrid.py 1 location


        resp.status = falcon.HTTP_200


class MicrogridSensorCollection:
    def __init__(self):
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_MICROGRID_ID')

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

        cursor.execute(" SELECT name "
                       " FROM tbl_microgrids "
                       " 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.MICROGRID_NOT_FOUND')

        query = (" SELECT s.id, s.name, s.uuid "
                 " FROM tbl_microgrids m, tbl_microgrids_sensors ms, tbl_sensors s "
                 " WHERE ms.microgrid_id = m.id AND s.id = ms.sensor_id AND m.id = %s "
                 " ORDER BY s.id ")
        cursor.execute(query, (id_,))
        rows = cursor.fetchall()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp, id_):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            print(str(ex))
            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_MICROGRID_ID')

        new_values = json.loads(raw_json)

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

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

        cursor.execute(" SELECT name "
                       " from tbl_microgrids "
                       " 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.MICROGRID_NOT_FOUND')

        cursor.execute(" SELECT name "
                       " FROM tbl_sensors "
                       " WHERE id = %s ", (sensor_id,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.SENSOR_NOT_FOUND')

        query = (" SELECT id "
                 " FROM tbl_microgrids_sensors "
                 " WHERE microgrid_id = %s AND sensor_id = %s")
        cursor.execute(query, (id_, sensor_id,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.MICROGRID_SENSOR_RELATION_EXISTS')

        add_row = (" INSERT INTO tbl_microgrids_sensors (microgrid_id, sensor_id) "
                   " VALUES (%s, %s) ")
        cursor.execute(add_row, (id_, sensor_id,))
        cnx.commit()
        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_201
        resp.location = '/microgrids/' + str(id_) + '/sensors/' + str(sensor_id)


class MicrogridSensorItem:
    def __init__(self):

		@@ 798-912 (lines=115) @@
795		resp.status = falcon.HTTP_200
796
797
798		class EnergyStoragePowerStationContainerCollection:
799		def __init__(self):
800		pass
801
802		@staticmethod
803		def on_options(req, resp, id_):
804		_ = req
805		resp.status = falcon.HTTP_200
806		_ = id_
807
808		@staticmethod
809		def on_get(req, resp, id_):
810		access_control(req)
811		if not id_.isdigit() or int(id_) <= 0:
812		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
813		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
814
815		cnx = mysql.connector.connect(**config.myems_system_db)
816		cursor = cnx.cursor()
817
818		cursor.execute(" SELECT name "
819		" FROM tbl_energy_storage_power_stations "
820		" WHERE id = %s ", (id_,))
821		if cursor.fetchone() is None:
822		cursor.close()
823		cnx.close()
824		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
825		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
826
827		query = (" SELECT s.id, s.name, s.uuid "
828		" FROM tbl_energy_storage_power_stations e, "
829		" tbl_energy_storage_power_stations_containers es, tbl_energy_storage_containers s "
830		" WHERE es.energy_storage_power_station_id = e.id "
831		" AND s.id = es.energy_storage_container_id "
832		" AND e.id = %s "
833		" ORDER BY s.id ")
834		cursor.execute(query, (id_,))
835		rows = cursor.fetchall()
836
837		result = list()
838		if rows is not None and len(rows) > 0:
839		for row in rows:
840		meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
841		result.append(meta_result)
842
843		resp.text = json.dumps(result)
844
845		@staticmethod
846		@user_logger
847		def on_post(req, resp, id_):
848		"""Handles POST requests"""
849		admin_control(req)
850		try:
851		raw_json = req.stream.read().decode('utf-8')
852		except Exception as ex:
853		print(str(ex))
854		raise falcon.HTTPError(status=falcon.HTTP_400,
855		title='API.BAD_REQUEST',
856		description='API.FAILED_TO_READ_REQUEST_STREAM')
857
858		if not id_.isdigit() or int(id_) <= 0:
859		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
860		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
861
862		new_values = json.loads(raw_json)
863
864		if 'energy_storage_container_id' not in new_values['data'].keys() or \
865		not isinstance(new_values['data']['energy_storage_container_id'], int) or \
866		new_values['data']['energy_storage_container_id'] <= 0:
867		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
868		description='API.INVALID_ENERGY_STORAGE_CONTAINER_ID')
869		energy_storage_container_id = new_values['data']['energy_storage_container_id']
870
871		cnx = mysql.connector.connect(**config.myems_system_db)
872		cursor = cnx.cursor()
873
874		cursor.execute(" SELECT name "
875		" FROM tbl_energy_storage_power_stations "
876		" WHERE id = %s ", (id_,))
877		if cursor.fetchone() is None:
878		cursor.close()
879		cnx.close()
880		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
881		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
882
883		cursor.execute(" SELECT name "
884		" FROM tbl_energy_storage_containers "
885		" WHERE id = %s ", (energy_storage_container_id,))
886		if cursor.fetchone() is None:
887		cursor.close()
888		cnx.close()
889		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
890		description='API.ENERGY_STORAGE_CONTAINER_NOT_FOUND')
891
892		query = (" SELECT id "
893		" FROM tbl_energy_storage_power_stations_containers "
894		" WHERE energy_storage_power_station_id = %s AND energy_storage_container_id = %s")
895		cursor.execute(query, (id_, energy_storage_container_id,))
896		if cursor.fetchone() is not None:
897		cursor.close()
898		cnx.close()
899		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
900		description='API.ENERGY_STORAGE_CONTAINER_SENSOR_RELATION_EXISTS')
901
902		add_row = (" INSERT INTO tbl_energy_storage_power_stations_containers "
903		" (energy_storage_power_station_id, energy_storage_container_id) "
904		" VALUES (%s, %s) ")
905		cursor.execute(add_row, (id_, energy_storage_container_id,))
906		cnx.commit()
907		cursor.close()
908		cnx.close()
909
910		resp.status = falcon.HTTP_201
911		resp.location = '/energystoragepowerstationss/' + str(id_) + '/containers/' + str(energy_storage_container_id)
912
913
914		class EnergyStoragePowerStationContainerItem:
915		def __init__(self):

		@@ 4882-4992 (lines=111) @@
4879		resp.status = falcon.HTTP_200
4880
4881
4882		class MicrogridSensorCollection:
4883		def __init__(self):
4884		pass
4885
4886		@staticmethod
4887		def on_options(req, resp, id_):
4888		_ = req
4889		resp.status = falcon.HTTP_200
4890		_ = id_
4891
4892		@staticmethod
4893		def on_get(req, resp, id_):
4894		access_control(req)
4895		if not id_.isdigit() or int(id_) <= 0:
4896		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4897		description='API.INVALID_MICROGRID_ID')
4898
4899		cnx = mysql.connector.connect(**config.myems_system_db)
4900		cursor = cnx.cursor()
4901
4902		cursor.execute(" SELECT name "
4903		" FROM tbl_microgrids "
4904		" WHERE id = %s ", (id_,))
4905		if cursor.fetchone() is None:
4906		cursor.close()
4907		cnx.close()
4908		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4909		description='API.MICROGRID_NOT_FOUND')
4910
4911		query = (" SELECT s.id, s.name, s.uuid "
4912		" FROM tbl_microgrids m, tbl_microgrids_sensors ms, tbl_sensors s "
4913		" WHERE ms.microgrid_id = m.id AND s.id = ms.sensor_id AND m.id = %s "
4914		" ORDER BY s.id ")
4915		cursor.execute(query, (id_,))
4916		rows = cursor.fetchall()
4917
4918		result = list()
4919		if rows is not None and len(rows) > 0:
4920		for row in rows:
4921		meta_result = {"id": row[0], "name": row[1], "uuid": row[2]}
4922		result.append(meta_result)
4923
4924		resp.text = json.dumps(result)
4925
4926		@staticmethod
4927		@user_logger
4928		def on_post(req, resp, id_):
4929		"""Handles POST requests"""
4930		admin_control(req)
4931		try:
4932		raw_json = req.stream.read().decode('utf-8')
4933		except Exception as ex:
4934		print(str(ex))
4935		raise falcon.HTTPError(status=falcon.HTTP_400,
4936		title='API.BAD_REQUEST',
4937		description='API.FAILED_TO_READ_REQUEST_STREAM')
4938
4939		if not id_.isdigit() or int(id_) <= 0:
4940		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4941		description='API.INVALID_MICROGRID_ID')
4942
4943		new_values = json.loads(raw_json)
4944
4945		if 'sensor_id' not in new_values['data'].keys() or \
4946		not isinstance(new_values['data']['sensor_id'], int) or \
4947		new_values['data']['sensor_id'] <= 0:
4948		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
4949		description='API.INVALID_SENSOR_ID')
4950		sensor_id = new_values['data']['sensor_id']
4951
4952		cnx = mysql.connector.connect(**config.myems_system_db)
4953		cursor = cnx.cursor()
4954
4955		cursor.execute(" SELECT name "
4956		" from tbl_microgrids "
4957		" WHERE id = %s ", (id_,))
4958		if cursor.fetchone() is None:
4959		cursor.close()
4960		cnx.close()
4961		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4962		description='API.MICROGRID_NOT_FOUND')
4963
4964		cursor.execute(" SELECT name "
4965		" FROM tbl_sensors "
4966		" WHERE id = %s ", (sensor_id,))
4967		if cursor.fetchone() is None:
4968		cursor.close()
4969		cnx.close()
4970		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
4971		description='API.SENSOR_NOT_FOUND')
4972
4973		query = (" SELECT id "
4974		" FROM tbl_microgrids_sensors "
4975		" WHERE microgrid_id = %s AND sensor_id = %s")
4976		cursor.execute(query, (id_, sensor_id,))
4977		if cursor.fetchone() is not None:
4978		cursor.close()
4979		cnx.close()
4980		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
4981		description='API.MICROGRID_SENSOR_RELATION_EXISTS')
4982
4983		add_row = (" INSERT INTO tbl_microgrids_sensors (microgrid_id, sensor_id) "
4984		" VALUES (%s, %s) ")
4985		cursor.execute(add_row, (id_, sensor_id,))
4986		cnx.commit()
4987		cursor.close()
4988		cnx.close()
4989
4990		resp.status = falcon.HTTP_201
4991		resp.location = '/microgrids/' + str(id_) + '/sensors/' + str(sensor_id)
4992
4993
4994		class MicrogridSensorItem:
4995		def __init__(self):

MyEMS / myems

Code Duplication Length = 111-115 lines in 2 locations

myems-api/core/energystoragepowerstation.py 1 location

myems-api/core/microgrid.py 1 location