core.apikey.ApiKeyCollection.on_options() - 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.apikey.ApiKeyCollection.on_options() A

↳ Parent: core.apikey

Complexity

Conditions

Size

Total Lines	11
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	1
nop	2

import os
import hashlib
from datetime import datetime, timezone, timedelta
import falcon
import mysql.connector
import simplejson as json
import config
from core.useractivity import admin_control


class ApiKeyCollection:
    """
    API Key Collection Resource

    This class handles API key management operations for the MyEMS system.
    It provides functionality to create, retrieve, and manage API keys
    used for programmatic access to the MyEMS API.
    """

    def __init__(self):
        pass

    @staticmethod
    def on_options(req, resp):
        """
        Handle OPTIONS request for CORS preflight

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all API keys

        Returns a list of all API keys with their metadata including:
        - API key ID and name
        - Token (hashed)
        - Creation datetime
        - Expiration datetime

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        admin_control(req)
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor()

        # Query to retrieve all API keys
        query = (" SELECT id, name, token, created_datetime_utc, expires_datetime_utc "
                 " FROM tbl_api_keys ")
        cursor.execute(query)
        rows = cursor.fetchall()

        # Build result list with timezone conversion
        token_list = 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:
                token_list.append({"id": row[0],
                                   "name": row[1],
                                   "token": row[2],
                                   "created_datetime": (row[3].replace(tzinfo=timezone.utc)
                                                        + timedelta(minutes=timezone_offset)).isoformat()[0:19],
                                   "expires_datetime": (row[4].replace(tzinfo=timezone.utc)
                                                        + timedelta(minutes=timezone_offset)).isoformat()[0:19]})

        cursor.close()
        cnx.close()
        resp.text = json.dumps(token_list)

    @staticmethod
    def on_post(req, resp):
        """
        Handle POST requests to create a new API key

        Creates a new API key with the specified name and expiration date.
        Generates a secure random token using SHA-512 hash.

        Args:
            req: Falcon request object containing API key data:
                - name: API key name (required)
                - expires_datetime: Expiration date (required)
            resp: Falcon response object
        """
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
            new_values = json.loads(raw_json)
        except UnicodeDecodeError as ex:
            print("Failed to decode request")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_ENCODING')
        except json.JSONDecodeError as ex:
            print("Failed to parse JSON")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_JSON_FORMAT')
        except Exception as ex:
            print("Unexpected error reading request stream")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')

        # Validate API key name
        if 'name' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['name'], str) or \
                len(str.strip(new_values['data']['name'])) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_API_KEY_NAME')
        name = str.strip(new_values['data']['name'])

        # Calculate timezone offset for datetime conversion
        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset

        # Parse and convert expiration datetime to UTC
        expires_datetime_local = datetime.strptime(new_values['data']['expires_datetime'], '%Y-%m-%dT%H:%M:%S')
        expires_datetime_utc = expires_datetime_local.replace(tzinfo=timezone.utc) - timedelta(minutes=timezone_offset)

        # Generate secure random token
        token = hashlib.sha512(os.urandom(16)).hexdigest()
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor()

        # Check if API key name already exists
        cursor.execute(" SELECT name FROM tbl_api_keys"
                       " WHERE name = %s ", (name,))
        rows = cursor.fetchall()

        if rows is not None and len(rows) > 0:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
                                   description='API.API_KEY_NAME_IS_ALREADY_IN_USE')

        # Insert new API key into database
        cursor.execute(" INSERT INTO tbl_api_keys "
                       " (name, token, created_datetime_utc, expires_datetime_utc) "
                       " VALUES(%s, %s, %s, %s) ", (name, token, datetime.utcnow(), expires_datetime_utc))

        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class ApiKeyItem:
    """
    API Key Item Resource

    This class handles individual API key operations including:
    - Retrieving a specific API key by ID
    - Updating API key information
    - Deleting API keys
    """

    def __init__(self):
        pass

    @staticmethod
    def on_options(req, resp, id_):
        """
        Handle OPTIONS request for CORS preflight

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: API key ID parameter
        """
        _ = req
        resp.status = falcon.HTTP_200
        _ = id_

    @staticmethod
    def on_get(req, resp, id_):
        """
        Handle GET requests to retrieve a specific API key by ID

        Retrieves a single API key with its metadata including:
        - API key ID and name
        - Token (hashed)
        - Creation datetime
        - Expiration datetime

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: API key ID to retrieve
        """
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title="API.INVALID_API_KEY_ID")

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

        # Query to retrieve specific API key by ID
        query = (" SELECT id, name, token, created_datetime_utc, expires_datetime_utc "
                 " FROM tbl_api_keys "
                 " 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.API_KEY_NOT_FOUND')
        else:
            # Convert UTC datetime to local timezone
            timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
            if config.utc_offset[0] == '-':
                timezone_offset = -timezone_offset
            meta_result = {"id": row[0],
                           "name": row[1],
                           "token": row[2],
                           "created_datetime": (row[3].replace(tzinfo=timezone.utc) +
                                                timedelta(minutes=timezone_offset)).isoformat()[0:19],
                           "expires_datetime": (row[4].replace(tzinfo=timezone.utc) +
                                                timedelta(minutes=timezone_offset)).isoformat()[0:19]}

        resp.text = json.dumps(meta_result)

    @staticmethod
    def on_put(req, resp, id_):
        """
        Handle PUT requests to update an API key

        Updates an existing API key with new name and expiration date.
        The token itself cannot be changed - only metadata.

        Args:
            req: Falcon request object containing update data:
                - name: New API key name (required)
                - expires_datetime: New expiration date (required)
            resp: Falcon response object
            id_: API key ID to update
        """
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except UnicodeDecodeError as ex:
            print("Failed to decode request")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_ENCODING')
        except Exception as ex:
            print("Unexpected error reading request stream")
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')

        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title="API.INVALID_API_KEY_ID")

        new_values = json.loads(raw_json)

        # Validate API key name
        if 'name' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['name'], str) or \
                len(str.strip(new_values['data']['name'])) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_API_KEY_NAME')
        name = str.strip(new_values['data']['name'])

        # Validate expiration datetime
        if 'expires_datetime' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['expires_datetime'], str) or \
                len(str.strip(new_values['data']['expires_datetime'])) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_EXPIRES_DATETIME')
        expires_datetime_local = str.strip(new_values['data']['expires_datetime'])

        # Calculate timezone offset for datetime conversion
        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset

        # Parse and convert expiration datetime to UTC
        try:
            expires_datetime_local = datetime.strptime(expires_datetime_local, '%Y-%m-%dT%H:%M:%S')
            expires_datetime_utc = expires_datetime_local.replace(tzinfo=timezone.utc) - \
                timedelta(minutes=timezone_offset)
        except ValueError:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description="API.INVALID_EXPIRES_DATETIME")

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

        # Check if new name conflicts with existing API keys
        cursor.execute(" SELECT name "
                       " FROM tbl_api_keys "
                       " WHERE name = %s ", (name,))
        if cursor.fetchall() is not None and \
                len(cursor.fetchall()) > 0:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.API_KEY_NAME_IS_ALREADY_IN_USE')

        # Check if API key exists
        cursor.execute(" SELECT token "
                       " FROM tbl_api_keys "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.API_KEY_NOT_FOUND')

        # Update API key information
        cursor.execute(" UPDATE tbl_api_keys "
                       " SET name = %s, expires_datetime_utc = %s "
                       " WHERE id = %s ", (name, expires_datetime_utc, id_))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200

    @staticmethod
    def on_delete(req, resp, id_):
        """
        Handle DELETE requests to remove an API key

        Deletes the specified API key from the database.
        This action cannot be undone.

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: API key ID to delete
        """
        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_API_KEY_ID')

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

        # Check if API key exists before deletion
        cursor.execute(" SELECT token "
                       " FROM tbl_api_keys "
                       " WHERE id = %s ", (id_,))
        if cursor.fetchone() is None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.API_KEY_NOT_FOUND')

        # Delete the API key
        cursor.execute(" DELETE FROM tbl_api_keys WHERE id = %s ", (id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204


1			import os
2			import hashlib
3			from datetime import datetime, timezone, timedelta
4			import falcon
5			import mysql.connector
6			import simplejson as json
7			import config
8			from core.useractivity import admin_control
9
10
11			class ApiKeyCollection:
12			"""
13			API Key Collection Resource
14
15			This class handles API key management operations for the MyEMS system.
16			It provides functionality to create, retrieve, and manage API keys
17			used for programmatic access to the MyEMS API.
18			"""
19
20			def __init__(self):
21			pass
22
23			@staticmethod
24			def on_options(req, resp):
25			"""
26			Handle OPTIONS request for CORS preflight
27
28			Args:
29			req: Falcon request object
30			resp: Falcon response object
31			"""
32			_ = req
33			resp.status = falcon.HTTP_200
34
35			@staticmethod
36			def on_get(req, resp):
37			"""
38			Handle GET requests to retrieve all API keys
39
40			Returns a list of all API keys with their metadata including:
41			- API key ID and name
42			- Token (hashed)
43			- Creation datetime
44			- Expiration datetime
45
46			Args:
47			req: Falcon request object
48			resp: Falcon response object
49			"""
50			admin_control(req)
51			cnx = mysql.connector.connect(**config.myems_user_db)
52			cursor = cnx.cursor()
53
54			# Query to retrieve all API keys
55			query = (" SELECT id, name, token, created_datetime_utc, expires_datetime_utc "
56			" FROM tbl_api_keys ")
57			cursor.execute(query)
58			rows = cursor.fetchall()
59
60			# Build result list with timezone conversion
61			token_list = list()
62			if rows is not None and len(rows) > 0:
63			timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
64			if config.utc_offset[0] == '-':
65			timezone_offset = -timezone_offset
66			for row in rows:
67			token_list.append({"id": row[0],
68			"name": row[1],
69			"token": row[2],
70			"created_datetime": (row[3].replace(tzinfo=timezone.utc)
71			+ timedelta(minutes=timezone_offset)).isoformat()[0:19],
72			"expires_datetime": (row[4].replace(tzinfo=timezone.utc)
73			+ timedelta(minutes=timezone_offset)).isoformat()[0:19]})
74
75			cursor.close()
76			cnx.close()
77			resp.text = json.dumps(token_list)
78
79			@staticmethod
80			def on_post(req, resp):
81			"""
82			Handle POST requests to create a new API key
83
84			Creates a new API key with the specified name and expiration date.
85			Generates a secure random token using SHA-512 hash.
86
87			Args:
88			req: Falcon request object containing API key data:
89			- name: API key name (required)
90			- expires_datetime: Expiration date (required)
91			resp: Falcon response object
92			"""
93			admin_control(req)
94			try:
95			raw_json = req.stream.read().decode('utf-8')
96			new_values = json.loads(raw_json)
97			except UnicodeDecodeError as ex:
98			print("Failed to decode request")
99			raise falcon.HTTPError(status=falcon.HTTP_400,
100			title='API.BAD_REQUEST',
101			description='API.INVALID_ENCODING')
102			except json.JSONDecodeError as ex:
103			print("Failed to parse JSON")
104			raise falcon.HTTPError(status=falcon.HTTP_400,
105			title='API.BAD_REQUEST',
106			description='API.INVALID_JSON_FORMAT')
107			except Exception as ex:
108			print("Unexpected error reading request stream")
109			raise falcon.HTTPError(status=falcon.HTTP_400,
110			title='API.BAD_REQUEST',
111			description='API.FAILED_TO_READ_REQUEST_STREAM')
112
113			# Validate API key name
114			if 'name' not in new_values['data'].keys() or \
115			not isinstance(new_values['data']['name'], str) or \
116			len(str.strip(new_values['data']['name'])) == 0:
117			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
118			description='API.INVALID_API_KEY_NAME')
119			name = str.strip(new_values['data']['name'])
120
121			# Calculate timezone offset for datetime conversion
122			timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
123			if config.utc_offset[0] == '-':
124			timezone_offset = -timezone_offset
125
126			# Parse and convert expiration datetime to UTC
127			expires_datetime_local = datetime.strptime(new_values['data']['expires_datetime'], '%Y-%m-%dT%H:%M:%S')
128			expires_datetime_utc = expires_datetime_local.replace(tzinfo=timezone.utc) - timedelta(minutes=timezone_offset)
129
130			# Generate secure random token
131			token = hashlib.sha512(os.urandom(16)).hexdigest()
132			cnx = mysql.connector.connect(**config.myems_user_db)
133			cursor = cnx.cursor()
134
135			# Check if API key name already exists
136			cursor.execute(" SELECT name FROM tbl_api_keys"
137			" WHERE name = %s ", (name,))
138			rows = cursor.fetchall()
139
140			if rows is not None and len(rows) > 0:
141			cursor.close()
142			cnx.close()
143			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.ERROR',
144			description='API.API_KEY_NAME_IS_ALREADY_IN_USE')
145
146			# Insert new API key into database
147			cursor.execute(" INSERT INTO tbl_api_keys "
148			" (name, token, created_datetime_utc, expires_datetime_utc) "
149			" VALUES(%s, %s, %s, %s) ", (name, token, datetime.utcnow(), expires_datetime_utc))
150
151			new_id = cursor.lastrowid
152			cnx.commit()
153			cursor.close()
154			cnx.close()
155
156			resp.status = falcon.HTTP_201
157			resp.location = '/apikeys/' + str(new_id)
158
159
160			class ApiKeyItem:
161			"""
162			API Key Item Resource
163
164			This class handles individual API key operations including:
165			- Retrieving a specific API key by ID
166			- Updating API key information
167			- Deleting API keys
168			"""
169
170			def __init__(self):
171			pass
172
173			@staticmethod
174			def on_options(req, resp, id_):
175			"""
176			Handle OPTIONS request for CORS preflight
177
178			Args:
179			req: Falcon request object
180			resp: Falcon response object
181			id_: API key ID parameter
182			"""
183			_ = req
184			resp.status = falcon.HTTP_200
185			_ = id_
186
187			@staticmethod
188			def on_get(req, resp, id_):
189			"""
190			Handle GET requests to retrieve a specific API key by ID
191
192			Retrieves a single API key with its metadata including:
193			- API key ID and name
194			- Token (hashed)
195			- Creation datetime
196			- Expiration datetime
197
198			Args:
199			req: Falcon request object
200			resp: Falcon response object
201			id_: API key ID to retrieve
202			"""
203			admin_control(req)
204			if not id_.isdigit() or int(id_) <= 0:
205			raise falcon.HTTPError(status=falcon.HTTP_400,
206			title="API.INVALID_API_KEY_ID")
207
208			cnx = mysql.connector.connect(**config.myems_user_db)
209			cursor = cnx.cursor()
210
211			# Query to retrieve specific API key by ID
212			query = (" SELECT id, name, token, created_datetime_utc, expires_datetime_utc "
213			" FROM tbl_api_keys "
214			" WHERE id = %s ")
215			cursor.execute(query, (id_,))
216			row = cursor.fetchone()
217			cursor.close()
218			cnx.close()
219
220			if row is None:
221			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
222			description='API.API_KEY_NOT_FOUND')
223			else:
224			# Convert UTC datetime to local timezone
225			timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
226			if config.utc_offset[0] == '-':
227			timezone_offset = -timezone_offset
228			meta_result = {"id": row[0],
229			"name": row[1],
230			"token": row[2],
231			"created_datetime": (row[3].replace(tzinfo=timezone.utc) +
232			timedelta(minutes=timezone_offset)).isoformat()[0:19],
233			"expires_datetime": (row[4].replace(tzinfo=timezone.utc) +
234			timedelta(minutes=timezone_offset)).isoformat()[0:19]}
235
236			resp.text = json.dumps(meta_result)
237
238			@staticmethod
239			def on_put(req, resp, id_):
240			"""
241			Handle PUT requests to update an API key
242
243			Updates an existing API key with new name and expiration date.
244			The token itself cannot be changed - only metadata.
245
246			Args:
247			req: Falcon request object containing update data:
248			- name: New API key name (required)
249			- expires_datetime: New expiration date (required)
250			resp: Falcon response object
251			id_: API key ID to update
252			"""
253			admin_control(req)
254			try:
255			raw_json = req.stream.read().decode('utf-8')
256			except UnicodeDecodeError as ex:
257			print("Failed to decode request")
258			raise falcon.HTTPError(status=falcon.HTTP_400,
259			title='API.BAD_REQUEST',
260			description='API.INVALID_ENCODING')
261			except Exception as ex:
262			print("Unexpected error reading request stream")
263			raise falcon.HTTPError(status=falcon.HTTP_400,
264			title='API.BAD_REQUEST',
265			description='API.FAILED_TO_READ_REQUEST_STREAM')
266
267			if not id_.isdigit() or int(id_) <= 0:
268			raise falcon.HTTPError(status=falcon.HTTP_400,
269			title="API.INVALID_API_KEY_ID")
270
271			new_values = json.loads(raw_json)
272
273			# Validate API key name
274			if 'name' not in new_values['data'].keys() or \
275			not isinstance(new_values['data']['name'], str) or \
276			len(str.strip(new_values['data']['name'])) == 0:
277			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
278			description='API.INVALID_API_KEY_NAME')
279			name = str.strip(new_values['data']['name'])
280
281			# Validate expiration datetime
282			if 'expires_datetime' not in new_values['data'].keys() or \
283			not isinstance(new_values['data']['expires_datetime'], str) or \
284			len(str.strip(new_values['data']['expires_datetime'])) == 0:
285			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
286			description='API.INVALID_EXPIRES_DATETIME')
287			expires_datetime_local = str.strip(new_values['data']['expires_datetime'])
288
289			# Calculate timezone offset for datetime conversion
290			timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
291			if config.utc_offset[0] == '-':
292			timezone_offset = -timezone_offset
293
294			# Parse and convert expiration datetime to UTC
295			try:
296			expires_datetime_local = datetime.strptime(expires_datetime_local, '%Y-%m-%dT%H:%M:%S')
297			expires_datetime_utc = expires_datetime_local.replace(tzinfo=timezone.utc) - \
298			timedelta(minutes=timezone_offset)
299			except ValueError:
300			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
301			description="API.INVALID_EXPIRES_DATETIME")
302
303			cnx = mysql.connector.connect(**config.myems_user_db)
304			cursor = cnx.cursor()
305
306			# Check if new name conflicts with existing API keys
307			cursor.execute(" SELECT name "
308			" FROM tbl_api_keys "
309			" WHERE name = %s ", (name,))
310			if cursor.fetchall() is not None and \
311			len(cursor.fetchall()) > 0:
312			cursor.close()
313			cnx.close()
314			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
315			description='API.API_KEY_NAME_IS_ALREADY_IN_USE')
316
317			# Check if API key exists
318			cursor.execute(" SELECT token "
319			" FROM tbl_api_keys "
320			" WHERE id = %s ", (id_,))
321			if cursor.fetchone() is None:
322			cursor.close()
323			cnx.close()
324			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
325			description='API.API_KEY_NOT_FOUND')
326
327			# Update API key information
328			cursor.execute(" UPDATE tbl_api_keys "
329			" SET name = %s, expires_datetime_utc = %s "
330			" WHERE id = %s ", (name, expires_datetime_utc, id_))
331			cnx.commit()
332
333			cursor.close()
334			cnx.close()
335
336			resp.status = falcon.HTTP_200
337
338			@staticmethod
339			def on_delete(req, resp, id_):
340			"""
341			Handle DELETE requests to remove an API key
342
343			Deletes the specified API key from the database.
344			This action cannot be undone.
345
346			Args:
347			req: Falcon request object
348			resp: Falcon response object
349			id_: API key ID to delete
350			"""
351			admin_control(req)
352			if not id_.isdigit() or int(id_) <= 0:
353			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
354			description='API.INVALID_API_KEY_ID')
355
356			cnx = mysql.connector.connect(**config.myems_user_db)
357			cursor = cnx.cursor()
358
359			# Check if API key exists before deletion
360			cursor.execute(" SELECT token "
361			" FROM tbl_api_keys "
362			" WHERE id = %s ", (id_,))
363			if cursor.fetchone() is None:
364			cursor.close()
365			cnx.close()
366			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
367			description='API.API_KEY_NOT_FOUND')
368
369			# Delete the API key
370			cursor.execute(" DELETE FROM tbl_api_keys WHERE id = %s ", (id_,))
371			cnx.commit()
372
373			cursor.close()
374			cnx.close()
375
376			resp.status = falcon.HTTP_204
377

MyEMS / myems

Push — master ( cb176a...4b8481 )

core.apikey.ApiKeyCollection.on_options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like