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):

		@@ 5045-5107 (lines=63) @@
5042		resp.location = '/photovoltaicpowerstations/' + str(id_) + '/users/' + str(user_id)
5043
5044
5045		class PhotovoltaicPowerStationUserItem:
5046		def __init__(self):
5047		pass
5048
5049		@staticmethod
5050		def on_options(req, resp, id_, uid):
5051		_ = req
5052		resp.status = falcon.HTTP_200
5053		_ = id_
5054
5055		@staticmethod
5056		@user_logger
5057		def on_delete(req, resp, id_, uid):
5058		# todo Verify if the user is bound when deleting it
5059		admin_control(req)
5060		if not id_.isdigit() or int(id_) <= 0:
5061		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5062		description='API.INVALID_PHOTOVOLTAIC_POWER_STATION_ID')
5063
5064		if not uid.isdigit() or int(uid) <= 0:
5065		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5066		description='API.INVALID_USER_ID')
5067
5068		cnx = mysql.connector.connect(**config.myems_system_db)
5069		cursor = cnx.cursor()
5070		cursor.execute(" SELECT name "
5071		" FROM tbl_photovoltaic_power_stations "
5072		" WHERE id = %s ", (id_,))
5073		if cursor.fetchone() is None:
5074		cursor.close()
5075		cnx.close()
5076		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5077		description='API.PHOTOVOLTAIC_POWER_STATION_NOT_FOUND')
5078
5079		cnx_user = mysql.connector.connect(**config.myems_user_db)
5080		cursor_user = cnx_user.cursor()
5081		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
5082		if cursor_user.fetchone() is None:
5083		cursor_user.close()
5084		cnx_user.close()
5085		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5086		description='API.USER_NOT_FOUND')
5087
5088		cursor.execute(" SELECT id "
5089		" FROM tbl_photovoltaic_power_stations_users "
5090		" WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
5091		if cursor.fetchone() is None:
5092		cursor.close()
5093		cnx.close()
5094		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5095		description='API.PHOTOVOLTAIC_POWER_STATION_USER_RELATION_NOT_FOUND')
5096
5097		cursor.execute(" DELETE FROM tbl_photovoltaic_power_stations_users "
5098		" WHERE photovoltaic_power_station_id = %s AND user_id = %s ", (id_, uid))
5099		cnx.commit()
5100
5101		cursor.close()
5102		cnx.close()
5103		cursor_user.close()
5104		cnx_user.close()
5105
5106		resp.status = falcon.HTTP_204
5107
5108
5109		class PhotovoltaicPowerStationDataSourceCollection:
5110		def __init__(self):

		@@ 1153-1215 (lines=63) @@
1150		resp.location = '/energystoragepowerstations/' + str(id_) + '/users/' + str(user_id)
1151
1152
1153		class EnergyStoragePowerStationUserItem:
1154		def __init__(self):
1155		pass
1156
1157		@staticmethod
1158		def on_options(req, resp, id_, uid):
1159		_ = req
1160		resp.status = falcon.HTTP_200
1161		_ = id_
1162
1163		@staticmethod
1164		@user_logger
1165		def on_delete(req, resp, id_, uid):
1166		# todo Verify if the user is bound when deleting it
1167		admin_control(req)
1168		if not id_.isdigit() or int(id_) <= 0:
1169		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1170		description='API.INVALID_ENERGY_STORAGE_POWER_STATION_ID')
1171
1172		if not uid.isdigit() or int(uid) <= 0:
1173		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1174		description='API.INVALID_USER_ID')
1175
1176		cnx = mysql.connector.connect(**config.myems_system_db)
1177		cursor = cnx.cursor()
1178		cursor.execute(" SELECT name "
1179		" FROM tbl_energy_storage_power_stations "
1180		" WHERE id = %s ", (id_,))
1181		if cursor.fetchone() is None:
1182		cursor.close()
1183		cnx.close()
1184		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1185		description='API.ENERGY_STORAGE_POWER_STATION_NOT_FOUND')
1186
1187		cnx_user = mysql.connector.connect(**config.myems_user_db)
1188		cursor_user = cnx_user.cursor()
1189		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
1190		if cursor_user.fetchone() is None:
1191		cursor_user.close()
1192		cnx_user.close()
1193		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1194		description='API.USER_NOT_FOUND')
1195
1196		cursor.execute(" SELECT id "
1197		" FROM tbl_energy_storage_power_stations_users "
1198		" WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
1199		if cursor.fetchone() is None:
1200		cursor.close()
1201		cnx.close()
1202		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1203		description='API.ENERGY_STORAGE_POWER_STATION_USER_RELATION_NOT_FOUND')
1204
1205		cursor.execute(" DELETE FROM tbl_energy_storage_power_stations_users "
1206		" WHERE energy_storage_power_station_id = %s AND user_id = %s ", (id_, uid))
1207		cnx.commit()
1208
1209		cursor.close()
1210		cnx.close()
1211		cursor_user.close()
1212		cnx_user.close()
1213
1214		resp.status = falcon.HTTP_204
1215
1216
1217		class EnergyStoragePowerStationExport:
1218		def __init__(self):

		@@ 5278-5339 (lines=62) @@
5275		resp.location = '/microgrids/' + str(id_) + '/users/' + str(user_id)
5276
5277
5278		class MicrogridUserItem:
5279		def __init__(self):
5280		pass
5281
5282		@staticmethod
5283		def on_options(req, resp, id_, uid):
5284		_ = req
5285		resp.status = falcon.HTTP_200
5286		_ = id_
5287
5288		@staticmethod
5289		@user_logger
5290		def on_delete(req, resp, id_, uid):
5291		# todo Verify if the user is bound when deleting it
5292		admin_control(req)
5293		if not id_.isdigit() or int(id_) <= 0:
5294		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5295		description='API.INVALID_MICROGRID_ID')
5296
5297		if not uid.isdigit() or int(uid) <= 0:
5298		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
5299		description='API.INVALID_USER_ID')
5300
5301		cnx = mysql.connector.connect(**config.myems_system_db)
5302		cursor = cnx.cursor()
5303		cursor.execute(" SELECT name "
5304		" FROM tbl_microgrids "
5305		" WHERE id = %s ", (id_,))
5306		if cursor.fetchone() is None:
5307		cursor.close()
5308		cnx.close()
5309		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5310		description='API.MICROGRID_NOT_FOUND')
5311
5312		cnx_user = mysql.connector.connect(**config.myems_user_db)
5313		cursor_user = cnx_user.cursor()
5314		cursor_user.execute(" SELECT name FROM tbl_users WHERE id = %s ", (uid,))
5315		if cursor_user.fetchone() is None:
5316		cursor_user.close()
5317		cnx_user.close()
5318		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5319		description='API.USER_NOT_FOUND')
5320
5321		cursor.execute(" SELECT id "
5322		" FROM tbl_microgrids_users "
5323		" WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
5324		if cursor.fetchone() is None:
5325		cursor.close()
5326		cnx.close()
5327		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
5328		description='API.MICROGRID_USER_RELATION_NOT_FOUND')
5329
5330		cursor.execute(" DELETE FROM tbl_microgrids_users WHERE microgrid_id = %s AND user_id = %s ", (id_, uid))
5331		cnx.commit()
5332
5333		cursor.close()
5334		cnx.close()
5335		cursor_user.close()
5336		cnx_user.close()
5337
5338		resp.status = falcon.HTTP_204
5339
5340
5341		class MicrogridExport:
5342		def __init__(self):

		@@ 1590-1651 (lines=62) @@
1587		resp.location = '/spaces/' + str(id_) + '/meters/' + str(meter_id)
1588
1589
1590		class SpaceMeterItem:
1591		def __init__(self):
1592		pass
1593
1594		@staticmethod
1595		def on_options(req, resp, id_, mid):
1596		_ = req
1597		resp.status = falcon.HTTP_200
1598		_ = id_
1599
1600		@staticmethod
1601		@user_logger
1602		def on_delete(req, resp, id_, mid):
1603		admin_control(req)
1604		if not id_.isdigit() or int(id_) <= 0:
1605		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1606		description='API.INVALID_SPACE_ID')
1607
1608		if not mid.isdigit() or int(mid) <= 0:
1609		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1610		description='API.INVALID_METER_ID')
1611
1612		cnx = mysql.connector.connect(**config.myems_system_db)
1613		cursor = cnx.cursor()
1614
1615		cursor.execute(" SELECT name "
1616		" FROM tbl_spaces "
1617		" WHERE id = %s ", (id_,))
1618		if cursor.fetchone() is None:
1619		cursor.close()
1620		cnx.close()
1621		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1622		description='API.SPACE_NOT_FOUND')
1623
1624		cursor.execute(" SELECT name "
1625		" FROM tbl_meters "
1626		" WHERE id = %s ", (mid,))
1627		if cursor.fetchone() is None:
1628		cursor.close()
1629		cnx.close()
1630		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1631		description='API.METER_NOT_FOUND')
1632
1633		cursor.execute(" SELECT id "
1634		" FROM tbl_spaces_meters "
1635		" WHERE space_id = %s AND meter_id = %s ", (id_, mid))
1636		# use fetchall to avoid 'Unread result found' error in case there are duplicate rows
1637		rows = cursor.fetchall()
1638		if rows is None or len(rows) == 0:
1639		cursor.close()
1640		cnx.close()
1641		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1642		description='API.SPACE_METER_RELATION_NOT_FOUND')
1643
1644		cursor.execute(" DELETE FROM tbl_spaces_meters WHERE space_id = %s AND meter_id = %s ", (id_, mid))
1645		cnx.commit()
1646
1647		cursor.close()
1648		cnx.close()
1649
1650		resp.status = falcon.HTTP_204
1651
1652
1653		class SpaceMicrogridCollection:
1654		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