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

Code Duplication Length = 78-82 lines in 3 locations

myems-api/core/emailmessage.py 1 location


        resp.text = json.dumps(result)


class EmailMessageItem:
    @staticmethod
    def __init__():
        """"Initializes EmailMessageItem"""
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(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.disconnect()

        if row is None:
            raise falcon.HTTPError(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
    def on_delete(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_EMAIL_MESSAGE_ID')

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

        cursor.execute(" SELECT id "
                       " FROM tbl_email_messages "
                       " WHERE id = %s ", (id_,))
        row = cursor.fetchone()

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

        cursor.execute(" DELETE FROM tbl_email_messages WHERE id = %s ", (id_,))
        cnx.commit()

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

        resp.status = falcon.HTTP_204



myems-api/core/wechatmessage.py 1 location


        resp.text = json.dumps(result)


class WechatMessageItem:
    @staticmethod
    def __init__():
        """"Initializes WechatMessageItem"""
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(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.disconnect()

        if row is None:
            raise falcon.HTTPError(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
    def on_delete(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_WECHAT_MESSAGE_ID')

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

        cursor.execute(" SELECT id "
                       " FROM tbl_wechat_messages_outbox "
                       " WHERE id = %s ", (id_,))
        row = cursor.fetchone()

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

        cursor.execute(" DELETE FROM tbl_wechat_messages_outbox WHERE id = %s ", (id_,))
        cnx.commit()

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

        resp.status = falcon.HTTP_204


myems-api/core/textmessage.py 1 location


        resp.text = json.dumps(result)


class TextMessageItem:
    @staticmethod
    def __init__():
        """"Initializes TextMessageItem"""
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(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.disconnect()

        if row is None:
            raise falcon.HTTPError(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
    def on_delete(req, resp, id_):
        access_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_TEXT_MESSAGE_ID')

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

        cursor.execute(" SELECT id FROM tbl_text_messages_outbox WHERE id = %s ", (id_,))
        row = cursor.fetchone()

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

        cursor.execute(" DELETE FROM tbl_text_messages_outbox WHERE id = %s ", (id_,))
        cnx.commit()

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

        resp.status = falcon.HTTP_204


		@@ 96-177 (lines=82) @@
93		resp.text = json.dumps(result)
94
95
96		class EmailMessageItem:
97		@staticmethod
98		def __init__():
99		""""Initializes EmailMessageItem"""
100		pass
101
102		@staticmethod
103		def on_options(req, resp, id_):
104		resp.status = falcon.HTTP_200
105
106		@staticmethod
107		def on_get(req, resp, id_):
108		access_control(req)
109		if not id_.isdigit() or int(id_) <= 0:
110		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
111		description='API.INVALID_EMAIL_MESSAGE_ID')
112
113		cnx = mysql.connector.connect(**config.myems_fdd_db)
114		cursor = cnx.cursor()
115
116		query = (" SELECT id, recipient_name, recipient_email, "
117		" subject, message, attachment_file_name, "
118		" created_datetime_utc, scheduled_datetime_utc, status "
119		" FROM tbl_email_messages "
120		" WHERE id = %s ")
121		cursor.execute(query, (id_,))
122		row = cursor.fetchone()
123
124		if cursor:
125		cursor.close()
126		if cnx:
127		cnx.disconnect()
128
129		if row is None:
130		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
131		description='API.EMAIL_MESSAGE_NOT_FOUND')
132
133		result = {"id": row[0],
134		"recipient_name": row[1],
135		"recipient_email": row[2],
136		"subject": row[3],
137		"message": row[4].replace("<br>", ""),
138		"attachment_file_name": row[5],
139		"created_datetime": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
140		"scheduled_datetime": row[7].timestamp() * 1000 if isinstance(row[7], datetime) else None,
141		"status": row[8]}
142
143		resp.text = json.dumps(result)
144
145		@staticmethod
146		@user_logger
147		def on_delete(req, resp, id_):
148		access_control(req)
149		if not id_.isdigit() or int(id_) <= 0:
150		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
151		description='API.INVALID_EMAIL_MESSAGE_ID')
152
153		cnx = mysql.connector.connect(**config.myems_fdd_db)
154		cursor = cnx.cursor()
155
156		cursor.execute(" SELECT id "
157		" FROM tbl_email_messages "
158		" WHERE id = %s ", (id_,))
159		row = cursor.fetchone()
160
161		if row is None:
162		if cursor:
163		cursor.close()
164		if cnx:
165		cnx.disconnect()
166		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
167		description='API.EMAIL_MESSAGE_NOT_FOUND')
168
169		cursor.execute(" DELETE FROM tbl_email_messages WHERE id = %s ", (id_,))
170		cnx.commit()
171
172		if cursor:
173		cursor.close()
174		if cnx:
175		cnx.disconnect()
176
177		resp.status = falcon.HTTP_204
178
179

		@@ 94-175 (lines=82) @@
91		resp.text = json.dumps(result)
92
93
94		class WechatMessageItem:
95		@staticmethod
96		def __init__():
97		""""Initializes WechatMessageItem"""
98		pass
99
100		@staticmethod
101		def on_options(req, resp, id_):
102		resp.status = falcon.HTTP_200
103
104		@staticmethod
105		def on_get(req, resp, id_):
106		access_control(req)
107		if not id_.isdigit() or int(id_) <= 0:
108		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
109		description='API.INVALID_WECHAT_MESSAGE_ID')
110
111		cnx = mysql.connector.connect(**config.myems_fdd_db)
112		cursor = cnx.cursor()
113
114		query = (" SELECT id, recipient_name, recipient_openid, message_template_id, "
115		" message_data, created_datetime_utc, scheduled_datetime_utc, "
116		" acknowledge_code, status "
117		" FROM tbl_wechat_messages_outbox "
118		" WHERE id = %s ")
119		cursor.execute(query, (id_,))
120		row = cursor.fetchone()
121
122		if cursor:
123		cursor.close()
124		if cnx:
125		cnx.disconnect()
126
127		if row is None:
128		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
129		description='API.WECHAT_MESSAGE_NOT_FOUND')
130
131		result = {"id": row[0],
132		"recipient_name": row[1],
133		"recipient_openid": row[2],
134		"recipient_template_id": row[3],
135		"message_data": row[4],
136		"created_datetime_utc": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
137		"scheduled_datetime_utc": row[6].timestamp() * 1000 if isinstance(row[6], datetime) else None,
138		"acknowledge_code": row[7],
139		"status": row[8]}
140
141		resp.text = json.dumps(result)
142
143		@staticmethod
144		@user_logger
145		def on_delete(req, resp, id_):
146		access_control(req)
147		if not id_.isdigit() or int(id_) <= 0:
148		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
149		description='API.INVALID_WECHAT_MESSAGE_ID')
150
151		cnx = mysql.connector.connect(**config.myems_fdd_db)
152		cursor = cnx.cursor()
153
154		cursor.execute(" SELECT id "
155		" FROM tbl_wechat_messages_outbox "
156		" WHERE id = %s ", (id_,))
157		row = cursor.fetchone()
158
159		if row is None:
160		if cursor:
161		cursor.close()
162		if cnx:
163		cnx.disconnect()
164		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
165		description='API.WECHAT_MESSAGE_NOT_FOUND')
166
167		cursor.execute(" DELETE FROM tbl_wechat_messages_outbox WHERE id = %s ", (id_,))
168		cnx.commit()
169
170		if cursor:
171		cursor.close()
172		if cnx:
173		cnx.disconnect()
174
175		resp.status = falcon.HTTP_204
176

		@@ 92-169 (lines=78) @@
89		resp.text = json.dumps(result)
90
91
92		class TextMessageItem:
93		@staticmethod
94		def __init__():
95		""""Initializes TextMessageItem"""
96		pass
97
98		@staticmethod
99		def on_options(req, resp, id_):
100		resp.status = falcon.HTTP_200
101
102		@staticmethod
103		def on_get(req, resp, id_):
104		access_control(req)
105		if not id_.isdigit() or int(id_) <= 0:
106		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
107		description='API.INVALID_TEXT_MESSAGE_ID')
108
109		cnx = mysql.connector.connect(**config.myems_fdd_db)
110		cursor = cnx.cursor()
111
112		query = (" SELECT id, recipient_name, recipient_mobile, "
113		" message, created_datetime_utc, scheduled_datetime_utc, acknowledge_code, status "
114		" FROM tbl_text_messages_outbox "
115		" WHERE id = %s ")
116		cursor.execute(query, (id_,))
117		row = cursor.fetchone()
118
119		if cursor:
120		cursor.close()
121		if cnx:
122		cnx.disconnect()
123
124		if row is None:
125		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
126		description='API.TEXT_MESSAGE_NOT_FOUND')
127
128		result = {"id": row[0],
129		"recipient_name": row[1],
130		"recipient_mobile": row[2],
131		"message": row[3],
132		"created_datetime": row[4].timestamp() * 1000 if isinstance(row[4], datetime) else None,
133		"scheduled_datetime": row[5].timestamp() * 1000 if isinstance(row[5], datetime) else None,
134		"acknowledge_code": row[6],
135		"status": row[7]}
136
137		resp.text = json.dumps(result)
138
139		@staticmethod
140		@user_logger
141		def on_delete(req, resp, id_):
142		access_control(req)
143		if not id_.isdigit() or int(id_) <= 0:
144		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
145		description='API.INVALID_TEXT_MESSAGE_ID')
146
147		cnx = mysql.connector.connect(**config.myems_fdd_db)
148		cursor = cnx.cursor()
149
150		cursor.execute(" SELECT id FROM tbl_text_messages_outbox WHERE id = %s ", (id_,))
151		row = cursor.fetchone()
152
153		if row is None:
154		if cursor:
155		cursor.close()
156		if cnx:
157		cnx.disconnect()
158		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
159		description='API.TEXT_MESSAGE_NOT_FOUND')
160
161		cursor.execute(" DELETE FROM tbl_text_messages_outbox WHERE id = %s ", (id_,))
162		cnx.commit()
163
164		if cursor:
165		cursor.close()
166		if cnx:
167		cnx.disconnect()
168
169		resp.status = falcon.HTTP_204
170

MyEMS / myems

Code Duplication Length = 78-82 lines in 3 locations

myems-api/core/emailmessage.py 1 location

myems-api/core/wechatmessage.py 1 location

myems-api/core/textmessage.py 1 location