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

Passed

Push — master ( f8b1cd...9ee003 )

unknown

created 2026-01-11 02:37 UTC

KnowledgeFileCollection.on_get() F

↳ Parent: core.knowledgefile

Complexity

Conditions

Size

Total Lines	91
Code Lines	67

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	67
dl	0
loc	91
rs	1.8
c	0
b	0
f	0
cc	17
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
import redis
from core.useractivity import user_logger, admin_control, access_control, api_key_control
import config


def clear_knowledgefile_cache(knowledge_file_id=None):

    """
    Clear knowledge file-related cache after data modification

    Args:
        knowledge_file_id: Knowledge file ID (optional, for specific file cache)
    """
    # Check if Redis is enabled
    if not config.redis.get('is_enabled', False):
        return

    redis_client = None
    try:
        redis_client = redis.Redis(
            host=config.redis['host'],
            port=config.redis['port'],
            password=config.redis['password'] if config.redis['password'] else None,
            db=config.redis['db'],
            decode_responses=True,
            socket_connect_timeout=2,
            socket_timeout=2
        )
        redis_client.ping()

        # Clear knowledge file list cache
        list_cache_key_pattern = 'knowledgefile:list*'
        matching_keys = redis_client.keys(list_cache_key_pattern)
        if matching_keys:
            redis_client.delete(*matching_keys)

        # Clear specific knowledge file cache if knowledge_file_id is provided
        if knowledge_file_id:
            item_cache_key = f'knowledgefile:item:{knowledge_file_id}'
            redis_client.delete(item_cache_key)

    except Exception:
        # If cache clear fails, ignore and continue
        pass


class KnowledgeFileCollection:
    """
    Knowledge File Collection Resource

    This class handles CRUD operations for knowledge file collection.
    It provides endpoints for listing all knowledge files and creating new files.
    Knowledge files contain documentation and reference materials for the energy management system.
    """
    def __init__(self):
        """Initialize KnowledgeFileCollection"""
        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):
        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)

        # Redis cache key
        cache_key = 'knowledgefile:list'
        cache_expire = 28800  # 8 hours in seconds (long-term cache)

        # Try to get from Redis cache (only if Redis is enabled)
        redis_client = None
        if config.redis.get('is_enabled', False):
            try:
                redis_client = redis.Redis(
                    host=config.redis['host'],
                    port=config.redis['port'],
                    password=config.redis['password'] if config.redis['password'] else None,
                    db=config.redis['db'],
                    decode_responses=True,
                    socket_connect_timeout=2,
                    socket_timeout=2
                )
                redis_client.ping()
                cached_result = redis_client.get(cache_key)
                if cached_result:
                    resp.text = cached_result
                    return
            except Exception:
                # If Redis connection fails, continue to database query
                pass

        # Cache miss or Redis error - query database
        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)

        # Store result in Redis cache
        result_json = json.dumps(result)
        if redis_client:
            try:
                redis_client.setex(cache_key, cache_expire, result_json)
            except Exception:
                # If cache set fails, ignore and continue
                pass

        resp.text = result_json

    @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_KNOWLEDGE_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_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()

        # Clear cache after creating new knowledge file
        clear_knowledgefile_cache()

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


class KnowledgeFileItem:
    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_):
        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')

        # Redis cache key
        cache_key = f'knowledgefile:item:{id_}'
        cache_expire = 28800  # 8 hours in seconds (long-term cache)

        # Try to get from Redis cache (only if Redis is enabled)
        redis_client = None
        if config.redis.get('is_enabled', False):
            try:
                redis_client = redis.Redis(
                    host=config.redis['host'],
                    port=config.redis['port'],
                    password=config.redis['password'] if config.redis['password'] else None,
                    db=config.redis['db'],
                    decode_responses=True,
                    socket_connect_timeout=2,
                    socket_timeout=2
                )
                redis_client.ping()
                cached_result = redis_client.get(cache_key)
                if cached_result:
                    resp.text = cached_result
                    return
            except Exception:
                # If Redis connection fails, continue to database query
                pass

        # Cache miss or Redis error - query database
        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)}

        # Store result in Redis cache
        result_json = json.dumps(result)
        if redis_client:
            try:
                redis_client.setex(cache_key, cache_expire, result_json)
            except Exception:
                # If cache set fails, ignore and continue
                pass

        resp.text = result_json

    @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 OSError as ex:
            print("Failed to stream request")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.KNOWLEDGE_FILE_CANNOT_BE_REMOVED_FROM_DISK')
        except Exception as ex:
            print("Unexpected error reading request stream")
            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()

        # Clear cache after deleting knowledge file
        clear_knowledgefile_cache(knowledge_file_id=id_)

        resp.status = falcon.HTTP_204


class KnowledgeFileRestore:

    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_):
        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 OSError as ex:
            print("Failed to stream request")
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.FAILED_TO_RESTORE_KNOWLEDGE_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_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		import redis
10		from core.useractivity import user_logger, admin_control, access_control, api_key_control
11		import config
12
13
14	View Code Duplication	def clear_knowledgefile_cache(knowledge_file_id=None):
		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...
15		"""
16		Clear knowledge file-related cache after data modification
17
18		Args:
19		knowledge_file_id: Knowledge file ID (optional, for specific file cache)
20		"""
21		# Check if Redis is enabled
22		if not config.redis.get('is_enabled', False):
23		return
24
25		redis_client = None
26		try:
27		redis_client = redis.Redis(
28		host=config.redis['host'],
29		port=config.redis['port'],
30		password=config.redis['password'] if config.redis['password'] else None,
31		db=config.redis['db'],
32		decode_responses=True,
33		socket_connect_timeout=2,
34		socket_timeout=2
35		)
36		redis_client.ping()
37
38		# Clear knowledge file list cache
39		list_cache_key_pattern = 'knowledgefile:list*'
40		matching_keys = redis_client.keys(list_cache_key_pattern)
41		if matching_keys:
42		redis_client.delete(*matching_keys)
43
44		# Clear specific knowledge file cache if knowledge_file_id is provided
45		if knowledge_file_id:
46		item_cache_key = f'knowledgefile:item:{knowledge_file_id}'
47		redis_client.delete(item_cache_key)
48
49		except Exception:
50		# If cache clear fails, ignore and continue
51		pass
52
53
54		class KnowledgeFileCollection:
55		"""
56		Knowledge File Collection Resource
57
58		This class handles CRUD operations for knowledge file collection.
59		It provides endpoints for listing all knowledge files and creating new files.
60		Knowledge files contain documentation and reference materials for the energy management system.
61		"""
62		def __init__(self):
63		"""Initialize KnowledgeFileCollection"""
64		pass
65
66		@staticmethod
67		def on_options(req, resp):
68		"""Handle OPTIONS requests for CORS preflight"""
69		_ = req
70		resp.status = falcon.HTTP_200
71
72		@staticmethod
73		def on_get(req, resp):
74		if 'API-KEY' not in req.headers or \
75		not isinstance(req.headers['API-KEY'], str) or \
76		len(str.strip(req.headers['API-KEY'])) == 0:
77		access_control(req)
78		else:
79		api_key_control(req)
80
81		# Redis cache key
82		cache_key = 'knowledgefile:list'
83		cache_expire = 28800 # 8 hours in seconds (long-term cache)
84
85		# Try to get from Redis cache (only if Redis is enabled)
86		redis_client = None
87		if config.redis.get('is_enabled', False):
88		try:
89		redis_client = redis.Redis(
90		host=config.redis['host'],
91		port=config.redis['port'],
92		password=config.redis['password'] if config.redis['password'] else None,
93		db=config.redis['db'],
94		decode_responses=True,
95		socket_connect_timeout=2,
96		socket_timeout=2
97		)
98		redis_client.ping()
99		cached_result = redis_client.get(cache_key)
100		if cached_result:
101		resp.text = cached_result
102		return
103		except Exception:
104		# If Redis connection fails, continue to database query
105		pass
106
107		# Cache miss or Redis error - query database
108		cnx = mysql.connector.connect(**config.myems_user_db)
109		cursor = cnx.cursor()
110
111		query = (" SELECT uuid, display_name "
112		" FROM tbl_users ")
113		cursor.execute(query)
114		rows = cursor.fetchall()
115		cursor.close()
116		cnx.close()
117
118		user_dict = dict()
119		if rows is not None and len(rows) > 0:
120		for row in rows:
121		user_dict[row[0]] = row[1]
122
123		cnx = mysql.connector.connect(**config.myems_system_db)
124		cursor = cnx.cursor()
125
126		query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object"
127		" FROM tbl_knowledge_files "
128		" ORDER BY upload_datetime_utc desc ")
129		cursor.execute(query)
130		rows = cursor.fetchall()
131		cursor.close()
132		cnx.close()
133
134		result = list()
135		if rows is not None and len(rows) > 0:
136		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
137		if config.utc_offset[0] == '-':
138		timezone_offset = -timezone_offset
139		for row in rows:
140		# Base64 encode the bytes
141		# get the Base64 encoded data using human-readable characters.
142		meta_result = {"id": row[0],
143		"file_name": row[1],
144		"uuid": row[2],
145		"upload_datetime": (row[3].replace(tzinfo=None)
146		+ timedelta(minutes=timezone_offset)).isoformat()[0:19],
147		"user_display_name": user_dict.get(row[4], None),
148		"file_size_bytes": sys.getsizeof(row[5]),
149		"file_bytes_base64": (base64.b64encode(row[5])).decode('utf-8')
150		}
151		result.append(meta_result)
152
153		# Store result in Redis cache
154		result_json = json.dumps(result)
155		if redis_client:
156		try:
157		redis_client.setex(cache_key, cache_expire, result_json)
158		except Exception:
159		# If cache set fails, ignore and continue
160		pass
161
162		resp.text = result_json
163
164		@staticmethod
165		@user_logger
166		def on_post(req, resp):
167		"""Handles POST requests"""
168		admin_control(req)
169		try:
170		upload = req.get_param('file')
171		# Read upload file as binary
172		raw_blob = upload.file.read()
173		# Retrieve filename
174		filename = upload.filename
175		file_uuid = str(uuid.uuid4())
176
177		# Define file_path
178		file_path = os.path.join(config.upload_path, file_uuid)
179
180		# Write to a temporary file to prevent incomplete files from being used.
181		with open(file_path + '~', 'wb') as f:
182		f.write(raw_blob)
183
184		# Now that we know the file has been fully saved to disk move it into place.
185		os.rename(file_path + '~', file_path)
186		except OSError as ex:
187		print("Failed to stream request")
188		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
189		description='API.FAILED_TO_UPLOAD_KNOWLEDGE_FILE')
190		except Exception as ex:
191		print("Unexpected error reading request stream")
192		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
193		description='API.FAILED_TO_UPLOAD_KNOWLEDGE_FILE')
194
195		# Verify User Session
196		token = req.headers.get('TOKEN')
197		user_uuid = req.headers.get('USER-UUID')
198		if token is None:
199		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
200		description='API.TOKEN_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
201		if user_uuid is None:
202		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
203		description='API.USER_UUID_NOT_FOUND_IN_HEADERS_PLEASE_LOGIN')
204
205		cnx = mysql.connector.connect(**config.myems_user_db)
206		cursor = cnx.cursor()
207
208		query = (" SELECT utc_expires "
209		" FROM tbl_sessions "
210		" WHERE user_uuid = %s AND token = %s")
211		cursor.execute(query, (user_uuid, token,))
212		row = cursor.fetchone()
213
214		if row is None:
215		if cursor:
216		cursor.close()
217		if cnx:
218		cnx.close()
219		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
220		description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
221		else:
222		utc_expires = row[0]
223		if datetime.utcnow() > utc_expires:
224		if cursor:
225		cursor.close()
226		if cnx:
227		cnx.close()
228		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
229		description='API.USER_SESSION_TIMEOUT')
230
231		cursor.execute(" SELECT id "
232		" FROM tbl_users "
233		" WHERE uuid = %s ",
234		(user_uuid,))
235		row = cursor.fetchone()
236		if row is None:
237		if cursor:
238		cursor.close()
239		if cnx:
240		cnx.close()
241		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
242		description='API.INVALID_USER_PLEASE_RE_LOGIN')
243
244		cnx = mysql.connector.connect(**config.myems_system_db)
245		cursor = cnx.cursor()
246
247		add_values = (" INSERT INTO tbl_knowledge_files "
248		" (file_name, uuid, upload_datetime_utc, upload_user_uuid, file_object ) "
249		" VALUES (%s, %s, %s, %s, %s) ")
250		cursor.execute(add_values, (filename,
251		file_uuid,
252		datetime.utcnow(),
253		user_uuid,
254		raw_blob))
255		new_id = cursor.lastrowid
256		cnx.commit()
257		cursor.close()
258		cnx.close()
259
260		# Clear cache after creating new knowledge file
261		clear_knowledgefile_cache()
262
263		resp.status = falcon.HTTP_201
264		resp.location = '/knowledgefiles/' + str(new_id)
265
266
267		class KnowledgeFileItem:
268		def __init__(self):
269		pass
270
271		@staticmethod
272		def on_options(req, resp, id_):
273		_ = req
274		resp.status = falcon.HTTP_200
275		_ = id_
276
277		@staticmethod
278		def on_get(req, resp, id_):
279		if 'API-KEY' not in req.headers or \
280		not isinstance(req.headers['API-KEY'], str) or \
281		len(str.strip(req.headers['API-KEY'])) == 0:
282		access_control(req)
283		else:
284		api_key_control(req)
285		if not id_.isdigit() or int(id_) <= 0:
286		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
287		description='API.INVALID_KNOWLEDGE_FILE_ID')
288
289		# Redis cache key
290		cache_key = f'knowledgefile:item:{id_}'
291		cache_expire = 28800 # 8 hours in seconds (long-term cache)
292
293		# Try to get from Redis cache (only if Redis is enabled)
294		redis_client = None
295		if config.redis.get('is_enabled', False):
296		try:
297		redis_client = redis.Redis(
298		host=config.redis['host'],
299		port=config.redis['port'],
300		password=config.redis['password'] if config.redis['password'] else None,
301		db=config.redis['db'],
302		decode_responses=True,
303		socket_connect_timeout=2,
304		socket_timeout=2
305		)
306		redis_client.ping()
307		cached_result = redis_client.get(cache_key)
308		if cached_result:
309		resp.text = cached_result
310		return
311		except Exception:
312		# If Redis connection fails, continue to database query
313		pass
314
315		# Cache miss or Redis error - query database
316		cnx = mysql.connector.connect(**config.myems_user_db)
317		cursor = cnx.cursor()
318
319		query = (" SELECT uuid, display_name "
320		" FROM tbl_users ")
321		cursor.execute(query)
322		rows = cursor.fetchall()
323		cursor.close()
324		cnx.close()
325
326		user_dict = dict()
327		if rows is not None and len(rows) > 0:
328		for row in rows:
329		user_dict[row[0]] = row[1]
330
331		cnx = mysql.connector.connect(**config.myems_system_db)
332		cursor = cnx.cursor()
333
334		query = (" SELECT id, file_name, uuid, upload_datetime_utc, upload_user_uuid "
335		" FROM tbl_knowledge_files "
336		" WHERE id = %s ")
337		cursor.execute(query, (id_,))
338		row = cursor.fetchone()
339		cursor.close()
340		cnx.close()
341
342		if row is None:
343		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
344		description='API.KNOWLEDGE_FILE_NOT_FOUND')
345
346		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
347		if config.utc_offset[0] == '-':
348		timezone_offset = -timezone_offset
349
350		result = {"id": row[0],
351		"file_name": row[1],
352		"uuid": row[2],
353		"upload_datetime": (row[3].replace(tzinfo=timezone.utc)
354		+ timedelta(minutes=timezone_offset)).isoformat()[0:19],
355		"user_display_name": user_dict.get(row[4], None)}
356
357		# Store result in Redis cache
358		result_json = json.dumps(result)
359		if redis_client:
360		try:
361		redis_client.setex(cache_key, cache_expire, result_json)
362		except Exception:
363		# If cache set fails, ignore and continue
364		pass
365
366		resp.text = result_json
367
368		@staticmethod
369		@user_logger
370		def on_delete(req, resp, id_):
371		"""Handles DELETE requests"""
372		admin_control(req)
373		if not id_.isdigit() or int(id_) <= 0:
374		raise falcon.HTTPError(status=falcon.HTTP_400,
375		title='API.BAD_REQUEST',
376		description='API.INVALID_KNOWLEDGE_FILE_ID')
377
378		cnx = mysql.connector.connect(**config.myems_system_db)
379		cursor = cnx.cursor()
380
381		cursor.execute(" SELECT uuid "
382		" FROM tbl_knowledge_files "
383		" WHERE id = %s ", (id_,))
384		row = cursor.fetchone()
385		if row is None:
386		cursor.close()
387		cnx.close()
388		raise falcon.HTTPError(status=falcon.HTTP_404,
389		title='API.NOT_FOUND',
390		description='API.KNOWLEDGE_FILE_NOT_FOUND')
391
392		try:
393		file_uuid = row[0]
394		# Define file_path
395		file_path = os.path.join(config.upload_path, file_uuid)
396		# remove the file from disk
397		os.remove(file_path)
398		except OSError as ex:
399		print("Failed to stream request")
400		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
401		description='API.KNOWLEDGE_FILE_CANNOT_BE_REMOVED_FROM_DISK')
402		except Exception as ex:
403		print("Unexpected error reading request stream")
404		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
405		description='API.KNOWLEDGE_FILE_CANNOT_BE_REMOVED_FROM_DISK')
406
407		cursor.execute(" DELETE FROM tbl_knowledge_files WHERE id = %s ", (id_,))
408		cnx.commit()
409
410		cursor.close()
411		cnx.close()
412
413		# Clear cache after deleting knowledge file
414		clear_knowledgefile_cache(knowledge_file_id=id_)
415
416		resp.status = falcon.HTTP_204
417
418
419	View Code Duplication	class KnowledgeFileRestore:
		0 ignored issues – show Duplication introduced 2026-01-11 02:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
420		def __init__(self):
421		pass
422
423		@staticmethod
424		def on_options(req, resp, id_):
425		_ = req
426		resp.status = falcon.HTTP_200
427		_ = id_
428
429		@staticmethod
430		def on_get(req, resp, id_):
431		admin_control(req)
432		if not id_.isdigit() or int(id_) <= 0:
433		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
434		description='API.INVALID_KNOWLEDGE_FILE_ID')
435
436		cnx = mysql.connector.connect(**config.myems_system_db)
437		cursor = cnx.cursor()
438
439		query = (" SELECT uuid, file_object "
440		" FROM tbl_knowledge_files "
441		" WHERE id = %s ")
442		cursor.execute(query, (id_,))
443		row = cursor.fetchone()
444		cursor.close()
445		cnx.close()
446
447		if row is None:
448		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
449		description='API.KNOWLEDGE_FILE_NOT_FOUND')
450
451		result = {"uuid": row[0],
452		"file_object": row[1]}
453		try:
454		raw_blob = result["file_object"]
455		file_uuid = result["uuid"]
456
457		# Define file_path
458		file_path = os.path.join(config.upload_path, file_uuid)
459
460		# Write to a temporary file to prevent incomplete files from
461		# being used.
462		temp_file_path = file_path + '~'
463
464		with open(temp_file_path, 'wb') as f:
465		f.write(raw_blob)
466
467		# Now that we know the file has been fully saved to disk
468		# move it into place.
469		os.replace(temp_file_path, file_path)
470		except OSError as ex:
471		print("Failed to stream request")
472		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
473		description='API.FAILED_TO_RESTORE_KNOWLEDGE_FILE')
474		except Exception as ex:
475		print("Unexpected error reading request stream")
476		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
477		description='API.FAILED_TO_RESTORE_KNOWLEDGE_FILE')
478		resp.text = json.dumps('success')
479

MyEMS / myems

Push — master ( f8b1cd...9ee003 )

KnowledgeFileCollection.on_get() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like