core.knowledgefile.KnowledgeFileCollection.on_post() - Code Metrics - Inspection of "Merge branch 'develop'" - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b3cd43...26687d )

unknown

created 2025-04-08 14:47 UTC

KnowledgeFileCollection.on_post() F

↳ Parent: core.knowledgefile

Complexity

Conditions

Size

Total Lines	93
Code Lines	68

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	68
dl	0
loc	93
rs	3.5672
c	0
b	0
f	0
cc	14
nop	2

How to fix Long Method Complexity

import base64
import os
import sys
import uuid
from datetime import datetime, timezone, timedelta
import falcon
import mysql.connector
import simplejson as json
from core.useractivity import user_logger, admin_control, access_control, api_key_control
import config


class KnowledgeFileCollection:
    def __init__(self):
        """"Initializes KnowledgeFileCollection"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        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.close()

        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, file_object"
                 " FROM tbl_knowledge_files "
                 " ORDER BY upload_datetime_utc desc ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

        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
                # get the Base64 encoded data using human-readable characters.
                meta_result = {"id": row[0],
                               "file_name": row[1],
                               "uuid": row[2],
                               "upload_datetime": (row[3].replace(tzinfo=None)
                                                   + timedelta(minutes=timezone_offset)).isoformat()[0:19],
                               "user_display_name": user_dict.get(row[4], None),
                               "file_size_bytes": sys.getsizeof(row[5]),
                               "file_bytes_base64": (base64.b64encode(row[5])).decode('utf-8')
                               }
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp):
        """Handles POST requests"""
        admin_control(req)
        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.
            with open(file_path + '~', 'wb') as f:
                f.write(raw_blob)

            # Now that we know the file has been fully saved to disk move it into place.
            os.rename(file_path + '~', file_path)
        except Exception as ex:
            raise falcon.HTTPError(status=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(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.TOKEN_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
        if user_uuid is None:
            raise falcon.HTTPError(status=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.close()
            raise falcon.HTTPError(status=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.close()
                raise falcon.HTTPError(status=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.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')

        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.close()

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


class KnowledgeFileItem:
    def __init__(self):
        """"Initializes KnowledgeFileItem"""
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=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.close()

        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.close()

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

        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset

        result = {"id": row[0],
                  "file_name": row[1],
                  "uuid": row[2],
                  "upload_datetime": (row[3].replace(tzinfo=timezone.utc)
                                      + timedelta(minutes=timezone_offset)).isoformat()[0:19],
                  "user_display_name": user_dict.get(row[4], None)}
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_):
        """Handles DELETE requests"""
        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_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.close()
            raise falcon.HTTPError(status=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(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.KNOWLEDGE_FILE_CANNOT_BE_REMOVED_FROM_DISK')

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

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204


class KnowledgeFileRestore:

    def __init__(self):
        """"Initializes KnowledgeFileRestore"""
        pass

    @staticmethod
    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_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.close()

        if row is None:
            raise falcon.HTTPError(status=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 + '~'

            with open(temp_file_path, 'wb') as f:
                f.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(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_RESTORE_KNOWLEDGE_FILE')
        resp.text = json.dumps('success')


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

MyEMS / myems

Push — master ( b3cd43...26687d )

KnowledgeFileCollection.on_post() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like