core.protocol.ProtocolItem.on_put() - Code Metrics - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

core.protocol.ProtocolItem.on_put() F
last analyzed 2025-12-22 11:02 UTC

↳ Parent: core.protocol

Complexity

Conditions

Size

Total Lines	102
Code Lines	63

Duplication

Lines	102
Ratio	100 %

Importance

Changes

Metric	Value
eloc	63
dl	102
loc	102
rs	3.6
c	0
b	0
f	0
cc	14
nop	3

How to fix Long Method Complexity

from datetime import datetime, timedelta
import falcon
import mysql.connector
import simplejson as json
from core.useractivity import user_logger, admin_control, access_control, api_key_control
import config


class ProtocolCollection:
    """
    Protocol Collection Resource

    This class handles CRUD operations for protocol collection.
    It provides endpoints for listing all protocols and creating new protocols.
    Protocols define communication standards used by data sources.
    """
    def __init__(self):
        """Initialize ProtocolCollection"""
        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):
        """
        Handle GET requests to retrieve all protocols

        Returns a list of all protocols with their metadata including:
        - Protocol ID
        - Protocol name
        - Protocol code

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        # Check authentication - use API key if provided, otherwise use access control
        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)

        # Connect to database and execute query
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, code "
                 " FROM tbl_protocols "
                 " ORDER BY id ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

        # Build result list
        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "code": row[2]}
                result.append(meta_result)

        resp.text = json.dumps(result)

    @staticmethod

    @user_logger
    def on_post(req, resp):
        """
        Handle POST requests to create a new protocol

        Creates a new protocol with the specified name and code.
        Validates that both name and code are unique.

        Args:
            req: Falcon request object containing protocol data:
                - name: Protocol name (required)
                - code: Protocol code (required)
            resp: Falcon response object
        """
        admin_control(req)

        # Read and parse request body
        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')

        new_values = json.loads(raw_json)

        # Validate protocol 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_PROTOCOL_NAME')
        name = str.strip(new_values['data']['name'])

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

        # Connect to database
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Check if protocol name already exists
        cursor.execute(" SELECT name "
                       " FROM tbl_protocols "
                       " WHERE name = %s ", (name,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')

        # Check if protocol code already exists
        cursor.execute(" SELECT code "
                       " FROM tbl_protocols "
                       " WHERE code = %s ", (code,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')

        # Insert new protocol
        add_row = (" INSERT INTO tbl_protocols "
                   "     (name, code) "
                   " VALUES (%s, %s) ")

        cursor.execute(add_row, (name,
                                 code))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class ProtocolItem:
    """
    Protocol Item Resource

    This class handles individual protocol operations.
    It provides endpoints for retrieving, updating, and deleting specific protocols.
    """

    def __init__(self):
        """Initialize ProtocolItem"""
        pass

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

    @staticmethod

    def on_get(req, resp, id_):
        """
        Handle GET requests to retrieve a specific protocol

        Returns the protocol details for the specified ID.

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: Protocol ID to retrieve
        """
        # Check authentication - use API key if provided, otherwise use access control
        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)

        # Validate protocol ID
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PROTOCOL_ID')

        # Connect to database and execute query
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, code "
                 " FROM tbl_protocols "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        # Check if protocol exists
        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.PROTOCOL_NOT_FOUND')

        # Build result
        result = {"id": row[0],
                  "name": row[1],
                  "code": row[2]}
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_):
        """
        Handle DELETE requests to delete a specific protocol

        Deletes the protocol with the specified ID.
        Checks if the protocol is being used by any data sources before deletion.

        Args:
            req: Falcon request object
            resp: Falcon response object
            id_: Protocol ID to delete
        """
        admin_control(req)

        # Validate protocol ID
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(
                status=falcon.HTTP_400,
                title='API.BAD_REQUEST',
                description='API.INVALID_PROTOCOL_ID'
            )

        # Connect to database
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Check if protocol exists
        cursor.execute("SELECT name,code FROM tbl_protocols 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.PROTOCOL_NOT_FOUND'
            )

        # Check if this protocol is being used by any data sources
        code = row[1]
        cursor.execute(" SELECT name "
                       " FROM tbl_data_sources "
                       " WHERE protocol = %s "
                       " LIMIT 1 ",
                       (code,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.THERE_IS_RELATION_WITH_DATA_SOURCES')

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

        cursor.close()
        cnx.close()
        resp.status = falcon.HTTP_204



    @staticmethod

    @user_logger
    def on_put(req, resp, id_):
        """
        Handle PUT requests to update a specific protocol

        Updates the protocol with the specified ID.
        Validates that the new name and code are unique.

        Args:
            req: Falcon request object containing updated protocol data:
                - name: Updated protocol name (required)
                - code: Updated protocol code (required)
            resp: Falcon response object
            id_: Protocol ID to update
        """
        admin_control(req)

        # Read and parse request body
        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')

        # Validate protocol ID
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PROTOCOL_ID')

        new_values = json.loads(raw_json)

        # Validate protocol 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_PROTOCOL_NAME')
        name = str.strip(new_values['data']['name'])

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

        # Connect to database
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Check if protocol exists
        cursor.execute(" SELECT name "
                       " FROM tbl_protocols "
                       " 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.PROTOCOL_NOT_FOUND')

        # Check if new name already exists (excluding current protocol)
        cursor.execute(" SELECT name "
                       " FROM tbl_protocols "
                       " WHERE name = %s AND id != %s ", (name, id_))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')

        # Check if new code already exists (excluding current protocol)
        cursor.execute(" SELECT code "
                       " FROM tbl_protocols "
                       " WHERE code = %s AND id != %s ", (code, id_))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')

        # Update the protocol
        update_row = (" UPDATE tbl_protocols "
                      " SET name = %s, code = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (name,
                                    code,
                                    id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200





1		from datetime import datetime, timedelta
2		import falcon
3		import mysql.connector
4		import simplejson as json
5		from core.useractivity import user_logger, admin_control, access_control, api_key_control
6		import config
7
8
9		class ProtocolCollection:
10		"""
11		Protocol Collection Resource
12
13		This class handles CRUD operations for protocol collection.
14		It provides endpoints for listing all protocols and creating new protocols.
15		Protocols define communication standards used by data sources.
16		"""
17		def __init__(self):
18		"""Initialize ProtocolCollection"""
19		pass
20
21		@staticmethod
22		def on_options(req, resp):
23		"""Handle OPTIONS requests for CORS preflight"""
24		_ = req
25		resp.status = falcon.HTTP_200
26
27		@staticmethod
28		def on_get(req, resp):
29		"""
30		Handle GET requests to retrieve all protocols
31
32		Returns a list of all protocols with their metadata including:
33		- Protocol ID
34		- Protocol name
35		- Protocol code
36
37		Args:
38		req: Falcon request object
39		resp: Falcon response object
40		"""
41		# Check authentication - use API key if provided, otherwise use access control
42		if 'API-KEY' not in req.headers or \
43		not isinstance(req.headers['API-KEY'], str) or \
44		len(str.strip(req.headers['API-KEY'])) == 0:
45		access_control(req)
46		else:
47		api_key_control(req)
48
49		# Connect to database and execute query
50		cnx = mysql.connector.connect(**config.myems_system_db)
51		cursor = cnx.cursor()
52
53		query = (" SELECT id, name, code "
54		" FROM tbl_protocols "
55		" ORDER BY id ")
56		cursor.execute(query)
57		rows = cursor.fetchall()
58		cursor.close()
59		cnx.close()
60
61		# Build result list
62		result = list()
63		if rows is not None and len(rows) > 0:
64		for row in rows:
65		meta_result = {"id": row[0],
66		"name": row[1],
67		"code": row[2]}
68		result.append(meta_result)
69
70		resp.text = json.dumps(result)
71
72	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2025-08-21 02:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
73		@user_logger
74		def on_post(req, resp):
75		"""
76		Handle POST requests to create a new protocol
77
78		Creates a new protocol with the specified name and code.
79		Validates that both name and code are unique.
80
81		Args:
82		req: Falcon request object containing protocol data:
83		- name: Protocol name (required)
84		- code: Protocol code (required)
85		resp: Falcon response object
86		"""
87		admin_control(req)
88
89		# Read and parse request body
90		try:
91		raw_json = req.stream.read().decode('utf-8')
92		except UnicodeDecodeError as ex:
93		print("Failed to decode request")
94		raise falcon.HTTPError(status=falcon.HTTP_400,
95		title='API.BAD_REQUEST',
96		description='API.INVALID_ENCODING')
97		except Exception as ex:
98		print("Unexpected error reading request stream")
99		raise falcon.HTTPError(status=falcon.HTTP_400,
100		title='API.BAD_REQUEST',
101		description='API.FAILED_TO_READ_REQUEST_STREAM')
102
103		new_values = json.loads(raw_json)
104
105		# Validate protocol name
106		if 'name' not in new_values['data'].keys() or \
107		not isinstance(new_values['data']['name'], str) or \
108		len(str.strip(new_values['data']['name'])) == 0:
109		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
110		description='API.INVALID_PROTOCOL_NAME')
111		name = str.strip(new_values['data']['name'])
112
113		# Validate protocol code
114		if 'code' not in new_values['data'].keys() or \
115		not isinstance(new_values['data']['code'], str) or \
116		len(str.strip(new_values['data']['code'])) == 0:
117		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
118		description='API.INVALID_PROTOCOL_CODE')
119		code = str.strip(new_values['data']['code'])
120
121		# Connect to database
122		cnx = mysql.connector.connect(**config.myems_system_db)
123		cursor = cnx.cursor()
124
125		# Check if protocol name already exists
126		cursor.execute(" SELECT name "
127		" FROM tbl_protocols "
128		" WHERE name = %s ", (name,))
129		if cursor.fetchone() is not None:
130		cursor.close()
131		cnx.close()
132		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
133		description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
134
135		# Check if protocol code already exists
136		cursor.execute(" SELECT code "
137		" FROM tbl_protocols "
138		" WHERE code = %s ", (code,))
139		if cursor.fetchone() is not None:
140		cursor.close()
141		cnx.close()
142		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
143		description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
144
145		# Insert new protocol
146		add_row = (" INSERT INTO tbl_protocols "
147		" (name, code) "
148		" VALUES (%s, %s) ")
149
150		cursor.execute(add_row, (name,
151		code))
152		new_id = cursor.lastrowid
153		cnx.commit()
154		cursor.close()
155		cnx.close()
156
157		resp.status = falcon.HTTP_201
158		resp.location = '/protocols/' + str(new_id)
159
160
161		class ProtocolItem:
162		"""
163		Protocol Item Resource
164
165		This class handles individual protocol operations.
166		It provides endpoints for retrieving, updating, and deleting specific protocols.
167		"""
168
169		def __init__(self):
170		"""Initialize ProtocolItem"""
171		pass
172
173		@staticmethod
174		def on_options(req, resp, id_):
175		"""Handle OPTIONS requests for CORS preflight"""
176		_ = req
177		resp.status = falcon.HTTP_200
178		_ = id_
179
180	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2025-10-10 00:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
181		def on_get(req, resp, id_):
182		"""
183		Handle GET requests to retrieve a specific protocol
184
185		Returns the protocol details for the specified ID.
186
187		Args:
188		req: Falcon request object
189		resp: Falcon response object
190		id_: Protocol ID to retrieve
191		"""
192		# Check authentication - use API key if provided, otherwise use access control
193		if 'API-KEY' not in req.headers or \
194		not isinstance(req.headers['API-KEY'], str) or \
195		len(str.strip(req.headers['API-KEY'])) == 0:
196		access_control(req)
197		else:
198		api_key_control(req)
199
200		# Validate protocol ID
201		if not id_.isdigit() or int(id_) <= 0:
202		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
203		description='API.INVALID_PROTOCOL_ID')
204
205		# Connect to database and execute query
206		cnx = mysql.connector.connect(**config.myems_system_db)
207		cursor = cnx.cursor()
208
209		query = (" SELECT id, name, code "
210		" FROM tbl_protocols "
211		" WHERE id = %s ")
212		cursor.execute(query, (id_,))
213		row = cursor.fetchone()
214		cursor.close()
215		cnx.close()
216
217		# Check if protocol exists
218		if row is None:
219		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
220		description='API.PROTOCOL_NOT_FOUND')
221
222		# Build result
223		result = {"id": row[0],
224		"name": row[1],
225		"code": row[2]}
226		resp.text = json.dumps(result)
227
228		@staticmethod
229		@user_logger
230		def on_delete(req, resp, id_):
231		"""
232		Handle DELETE requests to delete a specific protocol
233
234		Deletes the protocol with the specified ID.
235		Checks if the protocol is being used by any data sources before deletion.
236
237		Args:
238		req: Falcon request object
239		resp: Falcon response object
240		id_: Protocol ID to delete
241		"""
242		admin_control(req)
243
244		# Validate protocol ID
245		if not id_.isdigit() or int(id_) <= 0:
246		raise falcon.HTTPError(
247		status=falcon.HTTP_400,
248		title='API.BAD_REQUEST',
249		description='API.INVALID_PROTOCOL_ID'
250		)
251
252		# Connect to database
253		cnx = mysql.connector.connect(**config.myems_system_db)
254		cursor = cnx.cursor()
255
256		# Check if protocol exists
257		cursor.execute("SELECT name,code FROM tbl_protocols WHERE id = %s", (id_,))
258		row = cursor.fetchone()
259		if row is None:
260		cursor.close()
261		cnx.close()
262		raise falcon.HTTPError(
263		status=falcon.HTTP_404,
264		title='API.NOT_FOUND',
265		description='API.PROTOCOL_NOT_FOUND'
266		)
267
268		# Check if this protocol is being used by any data sources
269		code = row[1]
270		cursor.execute(" SELECT name "
271		" FROM tbl_data_sources "
272		" WHERE protocol = %s "
273		" LIMIT 1 ",
274		(code,))
275		if cursor.fetchone() is not None:
276		cursor.close()
277		cnx.close()
278		raise falcon.HTTPError(status=falcon.HTTP_400,
279		title='API.BAD_REQUEST',
280		description='API.THERE_IS_RELATION_WITH_DATA_SOURCES')
281
282		# Delete the protocol
283		cursor.execute(" DELETE FROM tbl_protocols WHERE id = %s ", (id_,))
284		cnx.commit()
285
286		cursor.close()
287		cnx.close()
288		resp.status = falcon.HTTP_204
289
290
291
292	View Code Duplication	@staticmethod
		0 ignored issues – show Duplication introduced 2021-03-02 03:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
293		@user_logger
294		def on_put(req, resp, id_):
295		"""
296		Handle PUT requests to update a specific protocol
297
298		Updates the protocol with the specified ID.
299		Validates that the new name and code are unique.
300
301		Args:
302		req: Falcon request object containing updated protocol data:
303		- name: Updated protocol name (required)
304		- code: Updated protocol code (required)
305		resp: Falcon response object
306		id_: Protocol ID to update
307		"""
308		admin_control(req)
309
310		# Read and parse request body
311		try:
312		raw_json = req.stream.read().decode('utf-8')
313		except UnicodeDecodeError as ex:
314		print("Failed to decode request")
315		raise falcon.HTTPError(status=falcon.HTTP_400,
316		title='API.BAD_REQUEST',
317		description='API.INVALID_ENCODING')
318		except Exception as ex:
319		print("Unexpected error reading request stream")
320		raise falcon.HTTPError(status=falcon.HTTP_400,
321		title='API.BAD_REQUEST',
322		description='API.FAILED_TO_READ_REQUEST_STREAM')
323
324		# Validate protocol ID
325		if not id_.isdigit() or int(id_) <= 0:
326		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
327		description='API.INVALID_PROTOCOL_ID')
328
329		new_values = json.loads(raw_json)
330
331		# Validate protocol name
332		if 'name' not in new_values['data'].keys() or \
333		not isinstance(new_values['data']['name'], str) or \
334		len(str.strip(new_values['data']['name'])) == 0:
335		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
336		description='API.INVALID_PROTOCOL_NAME')
337		name = str.strip(new_values['data']['name'])
338
339		# Validate protocol code
340		if 'code' not in new_values['data'].keys() or \
341		not isinstance(new_values['data']['code'], str) or \
342		len(str.strip(new_values['data']['code'])) == 0:
343		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
344		description='API.INVALID_PROTOCOL_CODE')
345		code = str.strip(new_values['data']['code'])
346
347		# Connect to database
348		cnx = mysql.connector.connect(**config.myems_system_db)
349		cursor = cnx.cursor()
350
351		# Check if protocol exists
352		cursor.execute(" SELECT name "
353		" FROM tbl_protocols "
354		" WHERE id = %s ", (id_,))
355		if cursor.fetchone() is None:
356		cursor.close()
357		cnx.close()
358		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
359		description='API.PROTOCOL_NOT_FOUND')
360
361		# Check if new name already exists (excluding current protocol)
362		cursor.execute(" SELECT name "
363		" FROM tbl_protocols "
364		" WHERE name = %s AND id != %s ", (name, id_))
365		if cursor.fetchone() is not None:
366		cursor.close()
367		cnx.close()
368		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
369		description='API.PROTOCOL_NAME_IS_ALREADY_IN_USE')
370
371		# Check if new code already exists (excluding current protocol)
372		cursor.execute(" SELECT code "
373		" FROM tbl_protocols "
374		" WHERE code = %s AND id != %s ", (code, id_))
375		if cursor.fetchone() is not None:
376		cursor.close()
377		cnx.close()
378		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
379		description='API.PROTOCOL_CODE_IS_ALREADY_IN_USE')
380
381		# Update the protocol
382		update_row = (" UPDATE tbl_protocols "
383		" SET name = %s, code = %s "
384		" WHERE id = %s ")
385		cursor.execute(update_row, (name,
386		code,
387		id_,))
388		cnx.commit()
389
390		cursor.close()
391		cnx.close()
392
393		resp.status = falcon.HTTP_200
394
395
396
397

MyEMS / myems

core.protocol.ProtocolItem.on_put() F last analyzed 2025-12-22 11:02 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

core.protocol.ProtocolItem.on_put() F
last analyzed 2025-12-22 11:02 UTC