core.knowledgefile - Code Metrics - Inspection of "Merge branch 'develop'" - myems/myems-api - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c9a013...90fdcc )

by Guangyu

created 2021-02-18 08:36 UTC

core.knowledgefile B

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	324
Duplicated Lines	16.98 %

Importance

Changes

Metric	Value
eloc	232
dl	55
loc	324
rs	8.8798
c	0
b	0
f	0
wmc	44

11 Methods

Rating	Name	Duplication	Size	Complexity
B	KnowledgeFileCollection.on_get()	0	54	8
A	KnowledgeFileCollection.on_options()	0	3	1
A	KnowledgeFileCollection.__init__()	0	3	1
A	KnowledgeFileRestore.__init__()	0	3	1
B	KnowledgeFileItem.on_delete()	39	39	5
D	KnowledgeFileCollection.on_post()	16	97	13
A	KnowledgeFileRestore.on_options()	0	3	1
B	KnowledgeFileRestore.on_get()	0	44	5
B	KnowledgeFileItem.on_get()	0	45	7
A	KnowledgeFileItem.on_options()	0	3	1
A	KnowledgeFileItem.__init__()	0	3	1

How to fix Duplicated Code Complexity

import falcon
import json
import mysql.connector
import config
import uuid
from datetime import datetime, timezone, timedelta
import os
import base64
import sys


class KnowledgeFileCollection:
    @staticmethod
    def __init__():
        pass

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

    @staticmethod
    def on_get(req, resp):
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor(dictionary=True)

        query = (" SELECT uuid, display_name "
                 " FROM tbl_users ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.disconnect()

        user_dict = dict()
        if rows is not None and len(rows) > 0:
            for row in rows:
                user_dict[row['uuid']] = row['display_name']

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

        query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object"
                 " FROM tbl_knowledge_files "
                 " ORDER BY upload_datetime_utc desc ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.disconnect()

        result = list()
        if rows is not None and len(rows) > 0:
            timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
            if config.utc_offset[0] == '-':
                timezone_offset = -timezone_offset
            for row in rows:
                # Base64 encode the bytes
                base64_encoded_data = base64.b64encode(row['file_object'])
                # get the Base64 encoded data using human-readable characters.
                base64_message = base64_encoded_data.decode('utf-8')
                upload_datetime_local = row['upload_datetime_utc'].replace(tzinfo=None) + \
                    timedelta(minutes=timezone_offset)
                upload_datetime = row['upload_datetime_utc']
                upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)
                meta_result = {"id": row['id'],
                               "file_name": row['file_name'],
                               "uuid": row['uuid'],
                               "upload_datetime": upload_datetime.timestamp() * 1000,
                               "upload_datetime_local": upload_datetime_local.isoformat(),
                               "user_display_name": user_dict.get(row['upload_user_uuid'], None),
                               "file_size_bytes": sys.getsizeof(row['file_object']),
                               "file_bytes_base64": base64_message
                               }
                result.append(meta_result)

        resp.body = json.dumps(result)

    @staticmethod
    def on_post(req, resp):
        """Handles POST requests"""

        try:
            upload = req.get_param('file')
            # Read upload file as binary
            raw_blob = upload.file.read()
            # Retrieve filename
            filename = upload.filename
            file_uuid = str(uuid.uuid4())

            # Define file_path
            file_path = os.path.join(config.upload_path, file_uuid)

            # Write to a temporary file to prevent incomplete files from
            # being used.
            temp_file_path = file_path + '~'

            open(temp_file_path, 'wb').write(raw_blob)

            # Now that we know the file has been fully saved to disk
            # move it into place.
            os.rename(temp_file_path, file_path)
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_UPLOAD_KNOWLEDGE_FILE')

        # Verify User Session
        token = req.headers.get('TOKEN')
        user_uuid = req.headers.get('USER-UUID')
        if token is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.TOKEN_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
        if user_uuid is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.USER_UUID_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')

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

        query = (" SELECT utc_expires "
                 " FROM tbl_sessions "
                 " WHERE user_uuid = %s AND token = %s")
        cursor.execute(query, (user_uuid, token,))
        row = cursor.fetchone()

        if row is None:

            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
        else:
            utc_expires = row[0]
            if datetime.utcnow() > utc_expires:
                if cursor:
                    cursor.close()
                if cnx:
                    cnx.disconnect()
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE uuid = %s ",
                       (user_uuid,))
        row = cursor.fetchone()
        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')
        else:
            user_id = row[0]

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

        add_values = (" INSERT INTO tbl_knowledge_files "
                      " (file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object ) "
                      " VALUES (%s, %s, %s, %s, %s) ")
        cursor.execute(add_values, (filename,
                                    file_uuid,
                                    datetime.utcnow(),
                                    user_uuid,
                                    raw_blob))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.disconnect()

        resp.status = falcon.HTTP_201
        resp.location = '/knowledgefiles/' + str(new_id)


class KnowledgeFileItem:
    @staticmethod
    def __init__():
        pass

    @staticmethod
    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_KNOWLEDGE_FILE_ID')

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

        query = (" SELECT uuid, display_name "
                 " FROM tbl_users ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.disconnect()

        user_dict = dict()
        if rows is not None and len(rows) > 0:
            for row in rows:
                user_dict[row[0]] = row[1]

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

        query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid "
                 " FROM tbl_knowledge_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.KNOWLEDGE_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,
                  "user_display_name": user_dict.get(row[4], None)}
        resp.body = json.dumps(result)

    @staticmethod

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

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

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

        try:
            file_uuid = row[0]
            # Define file_path
            file_path = os.path.join(config.upload_path, file_uuid)

            # remove the file from disk
            os.remove(file_path)
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
                                   description='API.KNOWLEDGE_FILE_NOT_FOUND')

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

        cursor.close()
        cnx.disconnect()

        resp.status = falcon.HTTP_204


class KnowledgeFileRestore:
    @staticmethod
    def __init__():
        pass

    @staticmethod
    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_KNOWLEDGE_FILE_ID')

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

        query = (" SELECT uuid, file_object "
                 " FROM tbl_knowledge_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.KNOWLEDGE_FILE_NOT_FOUND')

        result = {"uuid": row[0],
                  "file_object": row[1]}
        try:
            raw_blob = result["file_object"]
            file_uuid = result["uuid"]

            # Define file_path
            file_path = os.path.join(config.upload_path, file_uuid)

            # Write to a temporary file to prevent incomplete files from
            # being used.
            temp_file_path = file_path + '~'

            open(temp_file_path, 'wb').write(raw_blob)

            # Now that we know the file has been fully saved to disk
            # move it into place.
            os.replace(temp_file_path, file_path)
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_RESTORE_KNOWLEDGE_FILE')
        resp.body = 'success'
        resp.status = falcon.HTTP_200



1		import falcon
2		import json
3		import mysql.connector
4		import config
5		import uuid
6		from datetime import datetime, timezone, timedelta
7		import os
8		import base64
9		import sys
10
11
12		class KnowledgeFileCollection:
13		@staticmethod
14		def __init__():
15		pass
16
17		@staticmethod
18		def on_options(req, resp):
19		resp.status = falcon.HTTP_200
20
21		@staticmethod
22		def on_get(req, resp):
23		cnx = mysql.connector.connect(**config.myems_user_db)
24		cursor = cnx.cursor(dictionary=True)
25
26		query = (" SELECT uuid, display_name "
27		" FROM tbl_users ")
28		cursor.execute(query)
29		rows = cursor.fetchall()
30		cursor.close()
31		cnx.disconnect()
32
33		user_dict = dict()
34		if rows is not None and len(rows) > 0:
35		for row in rows:
36		user_dict[row['uuid']] = row['display_name']
37
38		cnx = mysql.connector.connect(**config.myems_system_db)
39		cursor = cnx.cursor(dictionary=True)
40
41		query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object"
42		" FROM tbl_knowledge_files "
43		" ORDER BY upload_datetime_utc desc ")
44		cursor.execute(query)
45		rows = cursor.fetchall()
46		cursor.close()
47		cnx.disconnect()
48
49		result = list()
50		if rows is not None and len(rows) > 0:
51		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
52		if config.utc_offset[0] == '-':
53		timezone_offset = -timezone_offset
54		for row in rows:
55		# Base64 encode the bytes
56		base64_encoded_data = base64.b64encode(row['file_object'])
57		# get the Base64 encoded data using human-readable characters.
58		base64_message = base64_encoded_data.decode('utf-8')
59		upload_datetime_local = row['upload_datetime_utc'].replace(tzinfo=None) + \
60		timedelta(minutes=timezone_offset)
61		upload_datetime = row['upload_datetime_utc']
62		upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)
63		meta_result = {"id": row['id'],
64		"file_name": row['file_name'],
65		"uuid": row['uuid'],
66		"upload_datetime": upload_datetime.timestamp() * 1000,
67		"upload_datetime_local": upload_datetime_local.isoformat(),
68		"user_display_name": user_dict.get(row['upload_user_uuid'], None),
69		"file_size_bytes": sys.getsizeof(row['file_object']),
70		"file_bytes_base64": base64_message
71		}
72		result.append(meta_result)
73
74		resp.body = json.dumps(result)
75
76		@staticmethod
77		def on_post(req, resp):
78		"""Handles POST requests"""
79
80		try:
81		upload = req.get_param('file')
82		# Read upload file as binary
83		raw_blob = upload.file.read()
84		# Retrieve filename
85		filename = upload.filename
86		file_uuid = str(uuid.uuid4())
87
88		# Define file_path
89		file_path = os.path.join(config.upload_path, file_uuid)
90
91		# Write to a temporary file to prevent incomplete files from
92		# being used.
93		temp_file_path = file_path + '~'
94
95		open(temp_file_path, 'wb').write(raw_blob)
96
97		# Now that we know the file has been fully saved to disk
98		# move it into place.
99		os.rename(temp_file_path, file_path)
100		except Exception as ex:
101		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
102		description='API.FAILED_TO_UPLOAD_KNOWLEDGE_FILE')
103
104		# Verify User Session
105		token = req.headers.get('TOKEN')
106		user_uuid = req.headers.get('USER-UUID')
107		if token is None:
108		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
109		description='API.TOKEN_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
110		if user_uuid is None:
111		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
112		description='API.USER_UUID_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
113
114		cnx = mysql.connector.connect(**config.myems_user_db)
115		cursor = cnx.cursor()
116
117		query = (" SELECT utc_expires "
118		" FROM tbl_sessions "
119		" WHERE user_uuid = %s AND token = %s")
120		cursor.execute(query, (user_uuid, token,))
121		row = cursor.fetchone()
122
123	View Code Duplication	if row is None:
		0 ignored issues – show Duplication introduced 2021-02-18 08:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
124		if cursor:
125		cursor.close()
126		if cnx:
127		cnx.disconnect()
128		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
129		description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
130		else:
131		utc_expires = row[0]
132		if datetime.utcnow() > utc_expires:
133		if cursor:
134		cursor.close()
135		if cnx:
136		cnx.disconnect()
137		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
138		description='API.USER_SESSION_TIMEOUT')
139
140		cursor.execute(" SELECT id "
141		" FROM tbl_users "
142		" WHERE uuid = %s ",
143		(user_uuid,))
144		row = cursor.fetchone()
145		if row is None:
146		if cursor:
147		cursor.close()
148		if cnx:
149		cnx.disconnect()
150		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
151		description='API.INVALID_USER_PLEASE_RE_LOGIN')
152		else:
153		user_id = row[0]
154
155		cnx = mysql.connector.connect(**config.myems_system_db)
156		cursor = cnx.cursor()
157
158		add_values = (" INSERT INTO tbl_knowledge_files "
159		" (file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object ) "
160		" VALUES (%s, %s, %s, %s, %s) ")
161		cursor.execute(add_values, (filename,
162		file_uuid,
163		datetime.utcnow(),
164		user_uuid,
165		raw_blob))
166		new_id = cursor.lastrowid
167		cnx.commit()
168		cursor.close()
169		cnx.disconnect()
170
171		resp.status = falcon.HTTP_201
172		resp.location = '/knowledgefiles/' + str(new_id)
173
174
175		class KnowledgeFileItem:
176		@staticmethod
177		def __init__():
178		pass
179
180		@staticmethod
181		def on_options(req, resp, id_):
182		resp.status = falcon.HTTP_200
183
184		@staticmethod
185		def on_get(req, resp, id_):
186		if not id_.isdigit() or int(id_) <= 0:
187		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
188		description='API.INVALID_KNOWLEDGE_FILE_ID')
189
190		cnx = mysql.connector.connect(**config.myems_user_db)
191		cursor = cnx.cursor()
192
193		query = (" SELECT uuid, display_name "
194		" FROM tbl_users ")
195		cursor.execute(query)
196		rows = cursor.fetchall()
197		cursor.close()
198		cnx.disconnect()
199
200		user_dict = dict()
201		if rows is not None and len(rows) > 0:
202		for row in rows:
203		user_dict[row[0]] = row[1]
204
205		cnx = mysql.connector.connect(**config.myems_system_db)
206		cursor = cnx.cursor()
207
208		query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid "
209		" FROM tbl_knowledge_files "
210		" WHERE id = %s ")
211		cursor.execute(query, (id_,))
212		row = cursor.fetchone()
213		cursor.close()
214		cnx.disconnect()
215
216		if row is None:
217		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
218		description='API.KNOWLEDGE_FILE_NOT_FOUND')
219
220		upload_datetime = row[3]
221		upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)
222
223		result = {"id": row[0],
224		"file_name": row[1],
225		"uuid": row[2],
226		"upload_datetime": upload_datetime.timestamp() * 1000,
227		"user_display_name": user_dict.get(row[4], None)}
228		resp.body = json.dumps(result)
229
230	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2020-11-24 10:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
231		def on_delete(req, resp, id_):
232		if not id_.isdigit() or int(id_) <= 0:
233		raise falcon.HTTPError(falcon.HTTP_400,
234		title='API.BAD_REQUEST',
235		description='API.INVALID_KNOWLEDGE_FILE_ID')
236
237		cnx = mysql.connector.connect(**config.myems_system_db)
238		cursor = cnx.cursor()
239
240		cursor.execute(" SELECT uuid "
241		" FROM tbl_knowledge_files "
242		" WHERE id = %s ", (id_,))
243		row = cursor.fetchone()
244		if row is None:
245		cursor.close()
246		cnx.disconnect()
247		raise falcon.HTTPError(falcon.HTTP_404,
248		title='API.NOT_FOUND',
249		description='API.KNOWLEDGE_FILE_NOT_FOUND')
250
251		try:
252		file_uuid = row[0]
253		# Define file_path
254		file_path = os.path.join(config.upload_path, file_uuid)
255
256		# remove the file from disk
257		os.remove(file_path)
258		except Exception as ex:
259		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
260		description='API.KNOWLEDGE_FILE_NOT_FOUND')
261
262		cursor.execute(" DELETE FROM tbl_knowledge_files WHERE id = %s ", (id_,))
263		cnx.commit()
264
265		cursor.close()
266		cnx.disconnect()
267
268		resp.status = falcon.HTTP_204
269
270
271		class KnowledgeFileRestore:
272		@staticmethod
273		def __init__():
274		pass
275
276		@staticmethod
277		def on_options(req, resp, id_):
278		resp.status = falcon.HTTP_200
279
280		@staticmethod
281		def on_get(req, resp, id_):
282		if not id_.isdigit() or int(id_) <= 0:
283		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
284		description='API.INVALID_KNOWLEDGE_FILE_ID')
285
286		cnx = mysql.connector.connect(**config.myems_system_db)
287		cursor = cnx.cursor()
288
289		query = (" SELECT uuid, file_object "
290		" FROM tbl_knowledge_files "
291		" WHERE id = %s ")
292		cursor.execute(query, (id_,))
293		row = cursor.fetchone()
294		cursor.close()
295		cnx.disconnect()
296
297		if row is None:
298		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
299		description='API.KNOWLEDGE_FILE_NOT_FOUND')
300
301		result = {"uuid": row[0],
302		"file_object": row[1]}
303		try:
304		raw_blob = result["file_object"]
305		file_uuid = result["uuid"]
306
307		# Define file_path
308		file_path = os.path.join(config.upload_path, file_uuid)
309
310		# Write to a temporary file to prevent incomplete files from
311		# being used.
312		temp_file_path = file_path + '~'
313
314		open(temp_file_path, 'wb').write(raw_blob)
315
316		# Now that we know the file has been fully saved to disk
317		# move it into place.
318		os.replace(temp_file_path, file_path)
319		except Exception as ex:
320		raise falcon.HTTPError(falcon.HTTP_400, title='API.ERROR',
321		description='API.FAILED_TO_RESTORE_KNOWLEDGE_FILE')
322		resp.body = 'success'
323		resp.status = falcon.HTTP_200
324
325

myems / myems-api

Push — master ( c9a013...90fdcc )

core.knowledgefile B

Complexity

Size/Duplication

Importance

11 Methods

How to fix Duplicated Code Complexity

Duplicated Code

Complexity

Duplication Side-by-Side

Filter issues like