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

Code Duplication Length = 36-38 lines in 4 locations

myems-api/core/wechatmessage.py 1 location


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

    @staticmethod
    def on_get(req, resp, id_):
        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_WECHAT_MESSAGE_ID')

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

        query = (" SELECT id, recipient_name, recipient_openid, message_template_id, "
                 "        message_data, created_datetime_utc, scheduled_datetime_utc, "
                 "        acknowledge_code, status "
                 " FROM tbl_wechat_messages_outbox "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if cursor:
            cursor.close()
        if cnx:
            cnx.close()

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

        result = {"id": row[0],
                  "recipient_name": row[1],
                  "recipient_openid": row[2],
                  "recipient_template_id": row[3],
                  "message_data": row[4],
                  "created_datetime_utc": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
                  "scheduled_datetime_utc": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
                  "acknowledge_code": row[7],
                  "status": row[8]}

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

myems-api/core/emailmessage.py 1 location


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

    @staticmethod
    def on_get(req, resp, id_):
        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_EMAIL_MESSAGE_ID')

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

        query = (" SELECT id, recipient_name, recipient_email, "
                 "        subject, message, attachment_file_name, "
                 "        created_datetime_utc, scheduled_datetime_utc, status "
                 " FROM tbl_email_messages "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if cursor:
            cursor.close()
        if cnx:
            cnx.close()

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

        result = {"id": row[0],
                  "recipient_name": row[1],
                  "recipient_email": row[2],
                  "subject": row[3],
                  "message": row[4].replace("<br>", ""),
                  "attachment_file_name": row[5],
                  "created_datetime": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
                  "scheduled_datetime": row[7].timestamp() * 1000 if isinstance(row[7], datetime) else None,
                  "status": row[8]}

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

myems-api/core/textmessage.py 1 location


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

    @staticmethod
    def on_get(req, resp, id_):
        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_TEXT_MESSAGE_ID')

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

        query = (" SELECT id, recipient_name, recipient_mobile, "
                 "        message, created_datetime_utc, scheduled_datetime_utc, acknowledge_code, status "
                 " FROM tbl_text_messages_outbox "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if cursor:
            cursor.close()
        if cnx:
            cnx.close()

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

        result = {"id": row[0],
                  "recipient_name": row[1],
                  "recipient_mobile": row[2],
                  "message": row[3],
                  "created_datetime": row[4].timestamp() * 1000 if isinstance(row[4], datetime) else None,
                  "scheduled_datetime": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
                  "acknowledge_code": row[6],
                  "status": row[7]}

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

myems-api/core/user.py 1 location


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

    @staticmethod
    def on_get(req, resp, id_):
        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_EMAIL_MESSAGE_ID')

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

        query = (" SELECT id, recipient_name, recipient_email, "
                 "        subject, message, created_datetime_utc, "
                 "        scheduled_datetime_utc, status "
                 " FROM tbl_email_messages "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if cursor:
            cursor.close()
        if cnx:
            cnx.close()

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

        result = {"id": row[0],
                  "recipient_name": row[1],
                  "recipient_email": row[2],
                  "subject": row[3],
                  "message": row[4].replace("<br>", ""),
                  "created_datetime": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
                  "scheduled_datetime": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
                  "status": row[7]}

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

		@@ 262-299 (lines=38) @@
259		def on_options(req, resp, id_):
260		resp.status = falcon.HTTP_200
261
262		@staticmethod
263		def on_get(req, resp, id_):
264		admin_control(req)
265		if not id_.isdigit() or int(id_) <= 0:
266		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
267		description='API.INVALID_WECHAT_MESSAGE_ID')
268
269		cnx = mysql.connector.connect(**config.myems_fdd_db)
270		cursor = cnx.cursor()
271
272		query = (" SELECT id, recipient_name, recipient_openid, message_template_id, "
273		" message_data, created_datetime_utc, scheduled_datetime_utc, "
274		" acknowledge_code, status "
275		" FROM tbl_wechat_messages_outbox "
276		" WHERE id = %s ")
277		cursor.execute(query, (id_,))
278		row = cursor.fetchone()
279
280		if cursor:
281		cursor.close()
282		if cnx:
283		cnx.close()
284
285		if row is None:
286		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
287		description='API.WECHAT_MESSAGE_NOT_FOUND')
288
289		result = {"id": row[0],
290		"recipient_name": row[1],
291		"recipient_openid": row[2],
292		"recipient_template_id": row[3],
293		"message_data": row[4],
294		"created_datetime_utc": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
295		"scheduled_datetime_utc": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
296		"acknowledge_code": row[7],
297		"status": row[8]}
298
299		resp.text = json.dumps(result)
300
301		@staticmethod
302		@user_logger

		@@ 253-290 (lines=38) @@
250		def on_options(req, resp, id_):
251		resp.status = falcon.HTTP_200
252
253		@staticmethod
254		def on_get(req, resp, id_):
255		admin_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_EMAIL_MESSAGE_ID')
259
260		cnx = mysql.connector.connect(**config.myems_fdd_db)
261		cursor = cnx.cursor()
262
263		query = (" SELECT id, recipient_name, recipient_email, "
264		" subject, message, attachment_file_name, "
265		" created_datetime_utc, scheduled_datetime_utc, status "
266		" FROM tbl_email_messages "
267		" WHERE id = %s ")
268		cursor.execute(query, (id_,))
269		row = cursor.fetchone()
270
271		if cursor:
272		cursor.close()
273		if cnx:
274		cnx.close()
275
276		if row is None:
277		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
278		description='API.EMAIL_MESSAGE_NOT_FOUND')
279
280		result = {"id": row[0],
281		"recipient_name": row[1],
282		"recipient_email": row[2],
283		"subject": row[3],
284		"message": row[4].replace("<br>", ""),
285		"attachment_file_name": row[5],
286		"created_datetime": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
287		"scheduled_datetime": row[7].timestamp() * 1000 if isinstance(row[7], datetime) else None,
288		"status": row[8]}
289
290		resp.text = json.dumps(result)
291
292		@staticmethod
293		@user_logger

		@@ 237-272 (lines=36) @@
234		def on_options(req, resp, id_):
235		resp.status = falcon.HTTP_200
236
237		@staticmethod
238		def on_get(req, resp, id_):
239		admin_control(req)
240		if not id_.isdigit() or int(id_) <= 0:
241		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
242		description='API.INVALID_TEXT_MESSAGE_ID')
243
244		cnx = mysql.connector.connect(**config.myems_fdd_db)
245		cursor = cnx.cursor()
246
247		query = (" SELECT id, recipient_name, recipient_mobile, "
248		" message, created_datetime_utc, scheduled_datetime_utc, acknowledge_code, status "
249		" FROM tbl_text_messages_outbox "
250		" WHERE id = %s ")
251		cursor.execute(query, (id_,))
252		row = cursor.fetchone()
253
254		if cursor:
255		cursor.close()
256		if cnx:
257		cnx.close()
258
259		if row is None:
260		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
261		description='API.TEXT_MESSAGE_NOT_FOUND')
262
263		result = {"id": row[0],
264		"recipient_name": row[1],
265		"recipient_mobile": row[2],
266		"message": row[3],
267		"created_datetime": row[4].timestamp() * 1000 if isinstance(row[4], datetime) else None,
268		"scheduled_datetime": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
269		"acknowledge_code": row[6],
270		"status": row[7]}
271
272		resp.text = json.dumps(result)
273
274		@staticmethod
275		@user_logger

		@@ 1341-1377 (lines=37) @@
1338		def on_options(req, resp, id_):
1339		resp.status = falcon.HTTP_200
1340
1341		@staticmethod
1342		def on_get(req, resp, id_):
1343		admin_control(req)
1344		if not id_.isdigit() or int(id_) <= 0:
1345		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
1346		description='API.INVALID_EMAIL_MESSAGE_ID')
1347
1348		cnx = mysql.connector.connect(**config.myems_user_db)
1349		cursor = cnx.cursor()
1350
1351		query = (" SELECT id, recipient_name, recipient_email, "
1352		" subject, message, created_datetime_utc, "
1353		" scheduled_datetime_utc, status "
1354		" FROM tbl_email_messages "
1355		" WHERE id = %s ")
1356		cursor.execute(query, (id_,))
1357		row = cursor.fetchone()
1358
1359		if cursor:
1360		cursor.close()
1361		if cnx:
1362		cnx.close()
1363
1364		if row is None:
1365		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
1366		description='API.EMAIL_MESSAGE_NOT_FOUND')
1367
1368		result = {"id": row[0],
1369		"recipient_name": row[1],
1370		"recipient_email": row[2],
1371		"subject": row[3],
1372		"message": row[4].replace("<br>", ""),
1373		"created_datetime": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
1374		"scheduled_datetime": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
1375		"status": row[7]}
1376
1377		resp.text = json.dumps(result)
1378
1379		@staticmethod
1380		@user_logger

MyEMS / myems

Code Duplication Length = 36-38 lines in 4 locations

myems-api/core/wechatmessage.py 1 location

myems-api/core/emailmessage.py 1 location

myems-api/core/textmessage.py 1 location

myems-api/core/user.py 1 location