core.knowledgefile.KnowledgeFileItem.on_delete() - 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

core.knowledgefile.KnowledgeFileItem.on_delete() B

↳ Parent: core.knowledgefile

Complexity

Conditions

Size

Total Lines	41
Code Lines	31

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	31
dl	0
loc	41
rs	8.6693
c	0
b	0
f	0
cc	5
nop	3

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 )

core.knowledgefile.KnowledgeFileItem.on_delete() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like