core.knowledgefile.KnowledgeFileCollection.on_post() - Code Metrics - Inspection of "added user_logger decorators to all on_put, on_pos..." - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c1feee...f28d3d )

by Guangyu

created 2021-08-27 10:11 UTC

KnowledgeFileCollection.on_post() F

↳ Parent: core.knowledgefile

Complexity

Conditions

Size

Total Lines	98
Code Lines	68

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	68
dl	0
loc	98
rs	3.9872
c	0
b	0
f	0
cc	13
nop	2

How to fix Long Method Complexity

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


class KnowledgeFileCollection:
    @staticmethod
    def __init__():
        """"Initializes KnowledgeFileCollection"""
        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
    @user_logger
    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__():
        """"Initializes KnowledgeFileItem"""
        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
    @user_logger
    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__():
        """"Initializes KnowledgeFileRestore"""
        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 = json.dumps('success')



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

MyEMS / myems

Push — master ( c1feee...f28d3d )

KnowledgeFileCollection.on_post() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like