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

Passed

Push — master ( cb176a...4b8481 )

unknown

created 2025-12-17 14:15 UTC

core.costfile.CostFileCollection.on_get() B

↳ Parent: core.costfile

Complexity

Conditions

Size

Total Lines	31
Code Lines	24

Duplication

Lines	31
Ratio	100 %

Importance

Changes

Metric	Value
eloc	24
dl	31
loc	31
rs	8.8373
c	0
b	0
f	0
cc	5
nop	2

import os
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
import config


class CostFileCollection:

    """
    Cost File Collection Resource

    This class handles CRUD operations for cost file collection.
    It provides endpoints for listing all cost files and creating new files.
    Cost files contain cost data for the energy management system.
    """
    def __init__(self):
        """Initialize CostFileCollection"""
        pass

    @staticmethod
    def on_options(req, resp):
        """Handle OPTIONS requests for CORS preflight"""
        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """Handles GET requests"""
        admin_control(req)
        cnx = mysql.connector.connect(**config.myems_historical_db)
        cursor = cnx.cursor()

        query = (" SELECT id, file_name, uuid, upload_datetime_utc, status "
                 " FROM tbl_cost_files "
                 " ORDER BY upload_datetime_utc desc ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

        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 = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_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],
                               "status": row[4]}
                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 OSError as ex: 
            print("Failed to stream request")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_UPLOAD_COST_FILE')
        except Exception as ex:
            print("Unexpected error reading request stream")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_UPLOAD_COST_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_user_db = mysql.connector.connect(**config.myems_user_db)
        cursor_user_db = cnx_user_db.cursor()

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

        if row is None:
            if cursor_user_db:
                cursor_user_db.close()
            if cnx_user_db:
                cnx_user_db.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_user_db:
                    cursor_user_db.close()
                if cnx_user_db:
                    cnx_user_db.close()
                raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor_user_db.execute(" SELECT id "
                               " FROM tbl_users "
                               " WHERE uuid = %s ",
                               (user_uuid,))
        row = cursor_user_db.fetchone()
        if row is None:
            if cursor_user_db:
                cursor_user_db.close()
            if cnx_user_db:
                cnx_user_db.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')

        cnx_historical_db = mysql.connector.connect(**config.myems_historical_db)
        cursor_historical_db = cnx_historical_db.cursor()

        add_values = (" INSERT INTO tbl_cost_files "
                      " (file_name, uuid, upload_datetime_utc, status, file_object ) "
                      " VALUES (%s, %s, %s, %s, %s) ")
        cursor_historical_db.execute(add_values, (filename,
                                                  file_uuid,
                                                  datetime.utcnow(),
                                                  'new',
                                                  raw_blob))
        new_id = cursor_historical_db.lastrowid
        cnx_historical_db.commit()
        cursor_historical_db.close()
        cnx_historical_db.close()

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


class CostFileItem:

    def __init__(self):
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        """Handles GET 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_COST_FILE_ID')

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

        query = (" SELECT id, file_name, uuid, upload_datetime_utc, status "
                 " FROM tbl_cost_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.COST_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],
                  "status": row[4]}
        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_COST_FILE_ID')

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

        cursor.execute(" SELECT uuid "
                       " FROM tbl_cost_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.COST_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 OSError as ex: 
            print("Failed to stream request")
        except Exception as ex:
            print(str(ex))
            # ignore exception and don't return API.COST_FILE_NOT_FOUND error
            pass

        # Note: the energy data imported from the deleted file will not be deleted
        cursor.execute(" DELETE FROM tbl_cost_files WHERE id = %s ", (id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204


class CostFileRestore:

    def __init__(self):
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        """Handles GET 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_COST_FILE_ID')

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

        query = (" SELECT uuid, file_object "
                 " FROM tbl_cost_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.COST_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 OSError as ex: 
            print("Failed to stream request")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_RESTORE_COST_FILE')
        except Exception as ex:
            print("Unexpected error reading request stream")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_RESTORE_COST_FILE')
        resp.text = json.dumps('success')


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

MyEMS / myems

Push — master ( cb176a...4b8481 )

core.costfile.CostFileCollection.on_get() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like