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

Code Duplication Length = 62-63 lines in 4 locations

myems-api/core/photovoltaicpowerstation.py 1 location


        resp.location = '/photovoltaicpowerstations/' + str(id_) + '/users/' + str(user_id)


class PhotovoltaicPowerStationUserItem:
    def __init__(self):
        pass

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

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_, uid):
        # todo Verify if the user is bound when deleting it
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_ID')

        if not uid.isdigit() or int(uid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_ID')

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()
        cursor.execute(" SELECT name "
                       " FROM tbl_photovoltaic_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.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')

        cnx_user = mysql.connector.connect(**config.myems_user_db)
        cursor_user = cnx_user.cursor()
        cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
        if cursor_user.fetchone() is None:
            cursor_user.close()
            cnx_user.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.USER_NOT_FOUND')

        cursor.execute(" SELECT id "
                       " FROM tbl_photovoltaic_power_stations_users "
                       " WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PHOTOVOLTAIC_POWER_STATION_USER_RELATION_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_photovoltaic_power_stations_users "
                       " WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
        cnx.commit()

        cursor.close()
        cnx.close()
        cursor_user.close()
        cnx_user.close()

        resp.status = falcon.HTTP_204


class PhotovoltaicPowerStationDataSourceCollection:
    def __init__(self):

myems-api/core/energystoragepowerstation.py 1 location


        resp.location = '/energystoragepowerstations/' + str(id_) + '/users/' + str(user_id)


class EnergyStoragePowerStationUserItem:
    def __init__(self):
        pass

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

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_, uid):
        # todo Verify if the user is bound when deleting it
        admin_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')

        if not uid.isdigit() or int(uid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_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')

        cnx_user = mysql.connector.connect(**config.myems_user_db)
        cursor_user = cnx_user.cursor()
        cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
        if cursor_user.fetchone() is None:
            cursor_user.close()
            cnx_user.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.USER_NOT_FOUND')

        cursor.execute(" SELECT id "
                       " FROM tbl_energy_storage_power_stations_users "
                       " WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
        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_USER_RELATION_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_energy_storage_power_stations_users "
                       " WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
        cnx.commit()

        cursor.close()
        cnx.close()
        cursor_user.close()
        cnx_user.close()

        resp.status = falcon.HTTP_204


class EnergyStoragePowerStationExport:
    def __init__(self):

myems-api/core/microgrid.py 1 location


        resp.location = '/microgrids/' + str(id_) + '/users/' + str(user_id)


class MicrogridUserItem:
    def __init__(self):
        pass

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

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_, uid):
        # todo Verify if the user is bound when deleting it
        admin_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')

        if not uid.isdigit() or int(uid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_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')

        cnx_user = mysql.connector.connect(**config.myems_user_db)
        cursor_user = cnx_user.cursor()
        cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
        if cursor_user.fetchone() is None:
            cursor_user.close()
            cnx_user.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.USER_NOT_FOUND')

        cursor.execute(" SELECT id "
                       " FROM tbl_microgrids_users "
                       " WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.MICROGRID_USER_RELATION_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_microgrids_users WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
        cnx.commit()

        cursor.close()
        cnx.close()
        cursor_user.close()
        cnx_user.close()

        resp.status = falcon.HTTP_204


class MicrogridExport:
    def __init__(self):

myems-api/core/space.py 1 location


        resp.location = '/spaces/' + str(id_) + '/meters/' + str(meter_id)


class SpaceMeterItem:
    def __init__(self):
        pass

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

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_, mid):
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SPACE_ID')

        if not mid.isdigit() or int(mid) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_METER_ID')

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

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

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

        cursor.execute(" SELECT id "
                       " FROM tbl_spaces_meters "
                       " WHERE space_id = %s AND meter_id = %s ", (id_, mid))
        # use fetchall to avoid 'Unread result found' error in case there are duplicate rows
        rows = cursor.fetchall()
        if rows is None or len(rows) == 0:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.SPACE_METER_RELATION_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_spaces_meters WHERE space_id = %s AND meter_id = %s ", (id_, mid))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204


class SpaceMicrogridCollection:
    def __init__(self):

		@@ 4995-5057 (lines=63) @@
4992		resp.location = '/photovoltaicpowerstations/' + str(id_) + '/users/' + str(user_id)
4993
4994
4995		class PhotovoltaicPowerStationUserItem:
4996		def __init__(self):
4997		pass
4998
4999		@staticmethod
5000		def on_options(req, resp, id_, uid):
5001		_ = req
5002		resp.status = falcon.HTTP_200
5003		_ = id_
5004
5005		@staticmethod
5006		@user_logger
5007		def on_delete(req, resp, id_, uid):
5008		# todo Verify if the user is bound when deleting it
5009		admin_control(req)
5010		if not id_.isdigit() or int(id_) <= 0:
5011		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5012		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_ID')
5013
5014		if not uid.isdigit() or int(uid) <= 0:
5015		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5016		description='API.INVALID_USER_ID')
5017
5018		cnx = mysql.connector.connect(**config.myems_system_db)
5019		cursor = cnx.cursor()
5020		cursor.execute(" SELECT name "
5021		" FROM tbl_photovoltaic_power_stations "
5022		" WHERE id = %s ", (id_,))
5023		if cursor.fetchone() is None:
5024		cursor.close()
5025		cnx.close()
5026		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5027		description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
5028
5029		cnx_user = mysql.connector.connect(**config.myems_user_db)
5030		cursor_user = cnx_user.cursor()
5031		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
5032		if cursor_user.fetchone() is None:
5033		cursor_user.close()
5034		cnx_user.close()
5035		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5036		description='API.USER_NOT_FOUND')
5037
5038		cursor.execute(" SELECT id "
5039		" FROM tbl_photovoltaic_power_stations_users "
5040		" WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
5041		if cursor.fetchone() is None:
5042		cursor.close()
5043		cnx.close()
5044		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5045		description='API.PHOTOVOLTAIC_POWER_STATION_USER_RELATION_NOT_FOUND')
5046
5047		cursor.execute(" DELETE FROM tbl_photovoltaic_power_stations_users "
5048		" WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
5049		cnx.commit()
5050
5051		cursor.close()
5052		cnx.close()
5053		cursor_user.close()
5054		cnx_user.close()
5055
5056		resp.status = falcon.HTTP_204
5057
5058
5059		class PhotovoltaicPowerStationDataSourceCollection:
5060		def __init__(self):

		@@ 1133-1195 (lines=63) @@
1130		resp.location = '/energystoragepowerstations/' + str(id_) + '/users/' + str(user_id)
1131
1132
1133		class EnergyStoragePowerStationUserItem:
1134		def __init__(self):
1135		pass
1136
1137		@staticmethod
1138		def on_options(req, resp, id_, uid):
1139		_ = req
1140		resp.status = falcon.HTTP_200
1141		_ = id_
1142
1143		@staticmethod
1144		@user_logger
1145		def on_delete(req, resp, id_, uid):
1146		# todo Verify if the user is bound when deleting it
1147		admin_control(req)
1148		if not id_.isdigit() or int(id_) <= 0:
1149		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1150		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
1151
1152		if not uid.isdigit() or int(uid) <= 0:
1153		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1154		description='API.INVALID_USER_ID')
1155
1156		cnx = mysql.connector.connect(**config.myems_system_db)
1157		cursor = cnx.cursor()
1158		cursor.execute(" SELECT name "
1159		" FROM tbl_energy_storage_power_stations "
1160		" WHERE id = %s ", (id_,))
1161		if cursor.fetchone() is None:
1162		cursor.close()
1163		cnx.close()
1164		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1165		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
1166
1167		cnx_user = mysql.connector.connect(**config.myems_user_db)
1168		cursor_user = cnx_user.cursor()
1169		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
1170		if cursor_user.fetchone() is None:
1171		cursor_user.close()
1172		cnx_user.close()
1173		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1174		description='API.USER_NOT_FOUND')
1175
1176		cursor.execute(" SELECT id "
1177		" FROM tbl_energy_storage_power_stations_users "
1178		" WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
1179		if cursor.fetchone() is None:
1180		cursor.close()
1181		cnx.close()
1182		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1183		description='API.ENERGY_STORAGE_POWER_STATION_USER_RELATION_NOT_FOUND')
1184
1185		cursor.execute(" DELETE FROM tbl_energy_storage_power_stations_users "
1186		" WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
1187		cnx.commit()
1188
1189		cursor.close()
1190		cnx.close()
1191		cursor_user.close()
1192		cnx_user.close()
1193
1194		resp.status = falcon.HTTP_204
1195
1196
1197		class EnergyStoragePowerStationExport:
1198		def __init__(self):

		@@ 5168-5229 (lines=62) @@
5165		resp.location = '/microgrids/' + str(id_) + '/users/' + str(user_id)
5166
5167
5168		class MicrogridUserItem:
5169		def __init__(self):
5170		pass
5171
5172		@staticmethod
5173		def on_options(req, resp, id_, uid):
5174		_ = req
5175		resp.status = falcon.HTTP_200
5176		_ = id_
5177
5178		@staticmethod
5179		@user_logger
5180		def on_delete(req, resp, id_, uid):
5181		# todo Verify if the user is bound when deleting it
5182		admin_control(req)
5183		if not id_.isdigit() or int(id_) <= 0:
5184		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5185		description='API.INVALID_MICROGRID_ID')
5186
5187		if not uid.isdigit() or int(uid) <= 0:
5188		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5189		description='API.INVALID_USER_ID')
5190
5191		cnx = mysql.connector.connect(**config.myems_system_db)
5192		cursor = cnx.cursor()
5193		cursor.execute(" SELECT name "
5194		" FROM tbl_microgrids "
5195		" WHERE id = %s ", (id_,))
5196		if cursor.fetchone() is None:
5197		cursor.close()
5198		cnx.close()
5199		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5200		description='API.MICROGRID_NOT_FOUND')
5201
5202		cnx_user = mysql.connector.connect(**config.myems_user_db)
5203		cursor_user = cnx_user.cursor()
5204		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
5205		if cursor_user.fetchone() is None:
5206		cursor_user.close()
5207		cnx_user.close()
5208		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5209		description='API.USER_NOT_FOUND')
5210
5211		cursor.execute(" SELECT id "
5212		" FROM tbl_microgrids_users "
5213		" WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
5214		if cursor.fetchone() is None:
5215		cursor.close()
5216		cnx.close()
5217		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5218		description='API.MICROGRID_USER_RELATION_NOT_FOUND')
5219
5220		cursor.execute(" DELETE FROM tbl_microgrids_users WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
5221		cnx.commit()
5222
5223		cursor.close()
5224		cnx.close()
5225		cursor_user.close()
5226		cnx_user.close()
5227
5228		resp.status = falcon.HTTP_204
5229
5230
5231		class MicrogridExport:
5232		def __init__(self):

		@@ 1555-1616 (lines=62) @@
1552		resp.location = '/spaces/' + str(id_) + '/meters/' + str(meter_id)
1553
1554
1555		class SpaceMeterItem:
1556		def __init__(self):
1557		pass
1558
1559		@staticmethod
1560		def on_options(req, resp, id_, mid):
1561		_ = req
1562		resp.status = falcon.HTTP_200
1563		_ = id_
1564
1565		@staticmethod
1566		@user_logger
1567		def on_delete(req, resp, id_, mid):
1568		admin_control(req)
1569		if not id_.isdigit() or int(id_) <= 0:
1570		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1571		description='API.INVALID_SPACE_ID')
1572
1573		if not mid.isdigit() or int(mid) <= 0:
1574		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1575		description='API.INVALID_METER_ID')
1576
1577		cnx = mysql.connector.connect(**config.myems_system_db)
1578		cursor = cnx.cursor()
1579
1580		cursor.execute(" SELECT name "
1581		" FROM tbl_spaces "
1582		" WHERE id = %s ", (id_,))
1583		if cursor.fetchone() is None:
1584		cursor.close()
1585		cnx.close()
1586		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1587		description='API.SPACE_NOT_FOUND')
1588
1589		cursor.execute(" SELECT name "
1590		" FROM tbl_meters "
1591		" WHERE id = %s ", (mid,))
1592		if cursor.fetchone() is None:
1593		cursor.close()
1594		cnx.close()
1595		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1596		description='API.METER_NOT_FOUND')
1597
1598		cursor.execute(" SELECT id "
1599		" FROM tbl_spaces_meters "
1600		" WHERE space_id = %s AND meter_id = %s ", (id_, mid))
1601		# use fetchall to avoid 'Unread result found' error in case there are duplicate rows
1602		rows = cursor.fetchall()
1603		if rows is None or len(rows) == 0:
1604		cursor.close()
1605		cnx.close()
1606		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1607		description='API.SPACE_METER_RELATION_NOT_FOUND')
1608
1609		cursor.execute(" DELETE FROM tbl_spaces_meters WHERE space_id = %s AND meter_id = %s ", (id_, mid))
1610		cnx.commit()
1611
1612		cursor.close()
1613		cnx.close()
1614
1615		resp.status = falcon.HTTP_204
1616
1617
1618		class SpaceMicrogridCollection:
1619		def __init__(self):

MyEMS / myems

Code Duplication Length = 62-63 lines in 4 locations

myems-api/core/photovoltaicpowerstation.py 1 location

myems-api/core/energystoragepowerstation.py 1 location

myems-api/core/microgrid.py 1 location

myems-api/core/space.py 1 location