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


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

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

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

        query = (" SELECT id, name, uuid, "
                 "        svg, description "
                 " FROM tbl_distribution_systems "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.disconnect()

        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.DISTRIBUTION_SYSTEM_NOT_FOUND')
        else:
            meta_result = {"id": row['id'],
                           "name": row['name'],
                           "uuid": row['uuid'],
                           "svg": row['svg'],
                           "description": row['description']}

        resp.body = json.dumps(meta_result)

    @staticmethod
    def on_delete(req, resp, id_):


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

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

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

        query = (" SELECT id, name, uuid, email, phone, description "
                 " FROM tbl_contacts "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.disconnect()

        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.CONTACT_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "uuid": row[2],
                  "email": row[3],
                  "phone": row[4],
                  "description": row[5]}
        resp.body = json.dumps(result)

    @staticmethod
    def on_delete(req, resp, id_):


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

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

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

        query = (" SELECT id, name, uuid, unit_of_measure, kgce, kgco2e "
                 " FROM tbl_energy_categories "
                 " WHERE id =%s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.disconnect()
        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.ENERGY_CATEGORY_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "uuid": row[2],
                  "unit_of_measure": row[3],
                  "kgce": row[4],
                  "kgco2e": row[5]}
        resp.body = json.dumps(result)

    @staticmethod
    def on_delete(req, resp, id_):


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

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

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

        query = (" SELECT id, file_name, uuid, upload_datetime_utc, status "
                 " FROM tbl_offline_meter_files "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.disconnect()
        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.OFFLINE_METER_FILE_NOT_FOUND')

        upload_datetime = row[3]
        upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)

        result = {"id": row[0],
                  "file_name": row[1],
                  "uuid": row[2],
                  "upload_datetime": upload_datetime.timestamp() * 1000,
                  "status": row[4]}
        resp.body = json.dumps(result)

    @staticmethod
    def on_delete(req, resp, id_):

		@@ 112-140 (lines=29) @@
109		def on_options(req, resp, id_):
110		resp.status = falcon.HTTP_200
111
112		@staticmethod
113		def on_get(req, resp, id_):
114		if not id_.isdigit() or int(id_) <= 0:
115		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
116		description='API.INVALID_METER_ID')
117
118		cnx = mysql.connector.connect(**config.myems_system_db)
119		cursor = cnx.cursor(dictionary=True)
120
121		query = (" SELECT id, name, uuid, "
122		" svg, description "
123		" FROM tbl_distribution_systems "
124		" WHERE id = %s ")
125		cursor.execute(query, (id_,))
126		row = cursor.fetchone()
127		cursor.close()
128		cnx.disconnect()
129
130		if row is None:
131		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
132		description='API.DISTRIBUTION_SYSTEM_NOT_FOUND')
133		else:
134		meta_result = {"id": row['id'],
135		"name": row['name'],
136		"uuid": row['uuid'],
137		"svg": row['svg'],
138		"description": row['description']}
139
140		resp.body = json.dumps(meta_result)
141
142		@staticmethod
143		def on_delete(req, resp, id_):

		@@ 127-154 (lines=28) @@
124		def on_options(req, resp, id_):
125		resp.status = falcon.HTTP_200
126
127		@staticmethod
128		def on_get(req, resp, id_):
129		if not id_.isdigit() or int(id_) <= 0:
130		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
131		description='API.INVALID_CONTACT_ID')
132
133		cnx = mysql.connector.connect(**config.myems_system_db)
134		cursor = cnx.cursor()
135
136		query = (" SELECT id, name, uuid, email, phone, description "
137		" FROM tbl_contacts "
138		" WHERE id = %s ")
139		cursor.execute(query, (id_,))
140		row = cursor.fetchone()
141		cursor.close()
142		cnx.disconnect()
143
144		if row is None:
145		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
146		description='API.CONTACT_NOT_FOUND')
147
148		result = {"id": row[0],
149		"name": row[1],
150		"uuid": row[2],
151		"email": row[3],
152		"phone": row[4],
153		"description": row[5]}
154		resp.body = json.dumps(result)
155
156		@staticmethod
157		def on_delete(req, resp, id_):

		@@ 117-143 (lines=27) @@
114		def on_options(req, resp, id_):
115		resp.status = falcon.HTTP_200
116
117		@staticmethod
118		def on_get(req, resp, id_):
119		if not id_.isdigit() or int(id_) <= 0:
120		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
121		description='API.INVALID_ENERGY_CATEGORY_ID')
122
123		cnx = mysql.connector.connect(**config.myems_system_db)
124		cursor = cnx.cursor()
125
126		query = (" SELECT id, name, uuid, unit_of_measure, kgce, kgco2e "
127		" FROM tbl_energy_categories "
128		" WHERE id =%s ")
129		cursor.execute(query, (id_,))
130		row = cursor.fetchone()
131		cursor.close()
132		cnx.disconnect()
133		if row is None:
134		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
135		description='API.ENERGY_CATEGORY_NOT_FOUND')
136
137		result = {"id": row[0],
138		"name": row[1],
139		"uuid": row[2],
140		"unit_of_measure": row[3],
141		"kgce": row[4],
142		"kgco2e": row[5]}
143		resp.body = json.dumps(result)
144
145		@staticmethod
146		def on_delete(req, resp, id_):

		@@ 153-182 (lines=30) @@
150		def on_options(req, resp, id_):
151		resp.status = falcon.HTTP_200
152
153		@staticmethod
154		def on_get(req, resp, id_):
155		if not id_.isdigit() or int(id_) <= 0:
156		raise falcon.HTTPError(falcon.HTTP_400,
157		title='API.BAD_REQUEST',
158		description='API.INVALID_OFFLINE_METER_FILE_ID')
159
160		cnx = mysql.connector.connect(**config.myems_historical_db)
161		cursor = cnx.cursor()
162
163		query = (" SELECT id, file_name, uuid, upload_datetime_utc, status "
164		" FROM tbl_offline_meter_files "
165		" WHERE id = %s ")
166		cursor.execute(query, (id_,))
167		row = cursor.fetchone()
168		cursor.close()
169		cnx.disconnect()
170		if row is None:
171		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
172		description='API.OFFLINE_METER_FILE_NOT_FOUND')
173
174		upload_datetime = row[3]
175		upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)
176
177		result = {"id": row[0],
178		"file_name": row[1],
179		"uuid": row[2],
180		"upload_datetime": upload_datetime.timestamp() * 1000,
181		"status": row[4]}
182		resp.body = json.dumps(result)
183
184		@staticmethod
185		def on_delete(req, resp, id_):

Code Duplication Length = 27-30 lines in 4 locations

core/distributionsystem.py 1 location

core/contact.py 1 location

core/energycategory.py 1 location

core/offlinemeterfile.py 1 location

myems / myems-api

Code Duplication Length = 27-30 lines in 4 locations

core/distributionsystem.py 1 location

core/contact.py 1 location

core/energycategory.py 1 location

core/offlinemeterfile.py 1 location