Issues in knowledgefile.py - All Issues - Inspection of "implemented equipment reports" - myems/myems-api - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( edcf12...e07a89 )

by Guangyu

created 2020-11-26 08:37 UTC

knowledgefile.py (2 issues)

Labels

Duplication 2

Severity

Informational 2

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


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

        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 "
                 " 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:
            for row in rows:
                upload_datetime = row[3]
                upload_datetime = upload_datetime.replace(tzinfo=timezone.utc)
                meta_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)}
                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
        cookies = req.headers['SET-COOKIE'].split('=')
        if 'user_uuid' not in cookies or 'token' not in cookies:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COOKIES_PLEASE_RE_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, (cookies[1], cookies[3],))
        row = cursor.fetchone()

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

myems / myems-api

Push — master ( edcf12...e07a89 )

knowledgefile.py (2 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like