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

Passed

Push — master ( d2b0a0...60251a )

unknown

created 2025-04-07 14:31 UTC

core.controlmode.ControlModeItem.init() A

↳ Parent: core.controlmode

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import uuid
from datetime import datetime, timedelta, timezone
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 ControlModeCollection:
    def __init__(self):
        """"Initializes"""
        pass

    @staticmethod
    def on_options(req, resp):
        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)
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid, is_active "
                 " FROM tbl_control_modes "
                 " ORDER BY id ")
        cursor.execute(query)
        rows = cursor.fetchall()

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

        cursor.close()
        cnx.close()

        resp.text = json.dumps(result)

    @staticmethod
    @user_logger
    def on_post(req, resp):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')
        new_values = json.loads(raw_json)

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

        if 'is_active' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_active'], bool):
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_ACTIVE_VALUE')
        is_active = new_values['data']['is_active']

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

        cursor.execute(" SELECT name "
                       " FROM tbl_control_modes "
                       " 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.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')

        add_row = (" INSERT INTO tbl_control_modes "
                   "             (name, uuid, is_active ) "
                   " VALUES (%s, %s, %s) ")
        cursor.execute(add_row, (name,
                                 str(uuid.uuid4()),
                                 is_active))
        new_id = cursor.lastrowid
        cnx.commit()

        cursor.close()
        cnx.close()

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


class ControlModeItem:
    def __init__(self):
        """"Initializes"""
        pass

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

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

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

        query = (" SELECT id, name, uuid, is_active "
                 " FROM tbl_control_modes "
                 " WHERE id = %s ")
        cursor.execute(query, (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.CONTROL_MODE_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "uuid": row[2],
                  "is_active": bool(row[3])}

        cursor.close()
        cnx.close()

        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_CONTROL_MODE_ID')

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

        cursor.execute(" SELECT name "
                       " FROM tbl_control_modes "
                       " 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.CONTROL_MODE_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_control_modes WHERE id = %s ", (id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_204

    @staticmethod
    @user_logger
    def on_put(req, resp, id_):
        """Handles PUT requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            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.BAD_REQUEST',
                                   description='API.INVALID_CONTROL_MODE_ID')

        new_values = json.loads(raw_json)

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

        if 'is_active' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['is_active'], bool):
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_ACTIVE_VALUE')
        is_active = new_values['data']['is_active']

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

        # check if the control mode exists
        query = (" SELECT name " 
                 " FROM tbl_control_modes " 
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        cursor.fetchone()

        if cursor.rowcount != 1:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.CONTROL_MODE_NOT_FOUND')

        cursor.execute(" SELECT name "
                       " FROM tbl_control_modes "
                       " 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.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')

        # update control mode
        update_row = (" UPDATE tbl_control_modes "
                      " SET name = %s, is_active = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (name,
                                    is_active,
                                    id_,))
        cnx.commit()

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


class ControlModeExport:

    def __init__(self):
        """"Initializes"""
        pass

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

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

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

        query = (" SELECT id, name, uuid, is_active "
                 " FROM tbl_control_modes "
                 " WHERE id = %s ")
        cursor.execute(query, (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.CONTROL_MODE_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "uuid": row[2],
                  "is_active": bool(row[3])}

        cursor.close()
        cnx.close()

        resp.text = json.dumps(result)


class ControlModeImport:
    def __init__(self):
        """"Initializes"""
        pass

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

    @staticmethod
    @user_logger
    def on_post(req, resp):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.FAILED_TO_READ_REQUEST_STREAM')
        new_values = json.loads(raw_json)

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

        if 'is_active' not in new_values.keys() or \
                not isinstance(new_values['is_active'], bool):
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_IS_ACTIVE_VALUE')
        is_active = new_values['is_active']

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

        cursor.execute(" SELECT name "
                       " FROM tbl_control_modes "
                       " 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.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')

        add_row = (" INSERT INTO tbl_control_modes "
                   "             (name, uuid, is_active ) "
                   " VALUES (%s, %s, %s) ")
        cursor.execute(add_row, (name,
                                 str(uuid.uuid4()),
                                 is_active,))
        new_id = cursor.lastrowid
        cnx.commit()

        cursor.close()
        cnx.close()

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


class ControlModeClone:

    def __init__(self):
        """"Initializes"""
        pass

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

    @staticmethod
    @user_logger
    def on_post(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_CONTROL_MODE_ID')

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

        query = (" SELECT id, name, uuid, is_active "
                 " FROM tbl_control_modes "
                 " WHERE id = %s ")
        cursor.execute(query, (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.CONTROL_MODE_NOT_FOUND')

        result = {"id": row[0],
                  "name": row[1],
                  "uuid": row[2],
                  "is_active": bool(row[3])}

        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset
        new_name = (str.strip(result['name']) +
                    (datetime.utcnow() + timedelta(minutes=timezone_offset)).isoformat(sep='-', timespec='seconds'))
        add_row = (" INSERT INTO tbl_control_modes "
                   "             (name, uuid, is_active) "
                   " VALUES (%s, %s, %s) ")
        cursor.execute(add_row, (new_name,
                                 str(uuid.uuid4()),
                                 result['is_active']))
        new_id = cursor.lastrowid
        cnx.commit()

        cursor.close()
        cnx.close()

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


1		import uuid
2		from datetime import datetime, timedelta, timezone
3		import falcon
4		import mysql.connector
5		import simplejson as json
6		from core.useractivity import user_logger, admin_control, access_control, api_key_control
7		import config
8
9
10		class ControlModeCollection:
11		def __init__(self):
12		""""Initializes"""
13		pass
14
15		@staticmethod
16		def on_options(req, resp):
17		resp.status = falcon.HTTP_200
18
19		@staticmethod
20		def on_get(req, resp):
21		if 'API-KEY' not in req.headers or \
22		not isinstance(req.headers['API-KEY'], str) or \
23		len(str.strip(req.headers['API-KEY'])) == 0:
24		access_control(req)
25		else:
26		api_key_control(req)
27		cnx = mysql.connector.connect(**config.myems_system_db)
28		cursor = cnx.cursor()
29
30		query = (" SELECT id, name, uuid, is_active "
31		" FROM tbl_control_modes "
32		" ORDER BY id ")
33		cursor.execute(query)
34		rows = cursor.fetchall()
35
36		result = list()
37		if rows is not None and len(rows) > 0:
38		for row in rows:
39		meta_result = {"id": row[0],
40		"name": row[1],
41		"uuid": row[2],
42		"is_active": bool(row[3])}
43		result.append(meta_result)
44
45		cursor.close()
46		cnx.close()
47
48		resp.text = json.dumps(result)
49
50		@staticmethod
51		@user_logger
52		def on_post(req, resp):
53		"""Handles POST requests"""
54		admin_control(req)
55		try:
56		raw_json = req.stream.read().decode('utf-8')
57		except Exception as ex:
58		raise falcon.HTTPError(status=falcon.HTTP_400,
59		title='API.BAD_REQUEST',
60		description='API.FAILED_TO_READ_REQUEST_STREAM')
61		new_values = json.loads(raw_json)
62
63		if 'name' not in new_values['data'].keys() or \
64		not isinstance(new_values['data']['name'], str) or \
65		len(str.strip(new_values['data']['name'])) == 0:
66		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
67		description='API.INVALID_METER_NAME')
68		name = str.strip(new_values['data']['name'])
69
70		if 'is_active' not in new_values['data'].keys() or \
71		not isinstance(new_values['data']['is_active'], bool):
72		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
73		description='API.INVALID_IS_ACTIVE_VALUE')
74		is_active = new_values['data']['is_active']
75
76		cnx = mysql.connector.connect(**config.myems_system_db)
77		cursor = cnx.cursor()
78
79		cursor.execute(" SELECT name "
80		" FROM tbl_control_modes "
81		" WHERE name = %s ", (name,))
82		if cursor.fetchone() is not None:
83		cursor.close()
84		cnx.close()
85		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
86		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
87
88		add_row = (" INSERT INTO tbl_control_modes "
89		" (name, uuid, is_active ) "
90		" VALUES (%s, %s, %s) ")
91		cursor.execute(add_row, (name,
92		str(uuid.uuid4()),
93		is_active))
94		new_id = cursor.lastrowid
95		cnx.commit()
96
97		cursor.close()
98		cnx.close()
99
100		resp.status = falcon.HTTP_201
101		resp.location = '/controlmodes/' + str(new_id)
102
103
104		class ControlModeItem:
105		def __init__(self):
106		""""Initializes"""
107		pass
108
109		@staticmethod
110		def on_options(req, resp, id_):
111		resp.status = falcon.HTTP_200
112
113		@staticmethod
114		def on_get(req, resp, id_):
115		"""Handles GET requests"""
116		if 'API-KEY' not in req.headers or \
117		not isinstance(req.headers['API-KEY'], str) or \
118		len(str.strip(req.headers['API-KEY'])) == 0:
119		access_control(req)
120		else:
121		api_key_control(req)
122		if not id_.isdigit() or int(id_) <= 0:
123		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
124		description='API.INVALID_CONTROL_MODE_ID')
125
126		cnx = mysql.connector.connect(**config.myems_system_db)
127		cursor = cnx.cursor()
128
129		query = (" SELECT id, name, uuid, is_active "
130		" FROM tbl_control_modes "
131		" WHERE id = %s ")
132		cursor.execute(query, (id_,))
133		row = cursor.fetchone()
134		if row is None:
135		cursor.close()
136		cnx.close()
137		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
138		description='API.CONTROL_MODE_NOT_FOUND')
139
140		result = {"id": row[0],
141		"name": row[1],
142		"uuid": row[2],
143		"is_active": bool(row[3])}
144
145		cursor.close()
146		cnx.close()
147
148		resp.text = json.dumps(result)
149
150		@staticmethod
151		@user_logger
152		def on_delete(req, resp, id_):
153		"""Handles DELETE requests"""
154		admin_control(req)
155		if not id_.isdigit() or int(id_) <= 0:
156		raise falcon.HTTPError(status=falcon.HTTP_400,
157		title='API.BAD_REQUEST',
158		description='API.INVALID_CONTROL_MODE_ID')
159
160		cnx = mysql.connector.connect(**config.myems_system_db)
161		cursor = cnx.cursor()
162
163		cursor.execute(" SELECT name "
164		" FROM tbl_control_modes "
165		" WHERE id = %s ", (id_,))
166		if cursor.fetchone() is None:
167		cursor.close()
168		cnx.close()
169		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
170		description='API.CONTROL_MODE_NOT_FOUND')
171
172		cursor.execute(" DELETE FROM tbl_control_modes WHERE id = %s ", (id_,))
173		cnx.commit()
174
175		cursor.close()
176		cnx.close()
177
178		resp.status = falcon.HTTP_204
179
180		@staticmethod
181		@user_logger
182		def on_put(req, resp, id_):
183		"""Handles PUT requests"""
184		admin_control(req)
185		try:
186		raw_json = req.stream.read().decode('utf-8')
187		except Exception as ex:
188		raise falcon.HTTPError(status=falcon.HTTP_400,
189		title='API.BAD_REQUEST',
190		description='API.FAILED_TO_READ_REQUEST_STREAM')
191
192		if not id_.isdigit() or int(id_) <= 0:
193		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
194		description='API.INVALID_CONTROL_MODE_ID')
195
196		new_values = json.loads(raw_json)
197
198		if 'name' not in new_values['data'].keys() or \
199		not isinstance(new_values['data']['name'], str) or \
200		len(str.strip(new_values['data']['name'])) == 0:
201		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
202		description='API.INVALID_CONTROL_MODE_NAME')
203		name = str.strip(new_values['data']['name'])
204
205		if 'is_active' not in new_values['data'].keys() or \
206		not isinstance(new_values['data']['is_active'], bool):
207		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
208		description='API.INVALID_IS_ACTIVE_VALUE')
209		is_active = new_values['data']['is_active']
210
211		cnx = mysql.connector.connect(**config.myems_system_db)
212		cursor = cnx.cursor()
213
214		# check if the control mode exists
215		query = (" SELECT name "
216		" FROM tbl_control_modes "
217		" WHERE id = %s ")
218		cursor.execute(query, (id_,))
219		cursor.fetchone()
220
221		if cursor.rowcount != 1:
222		cursor.close()
223		cnx.close()
224		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
225		description='API.CONTROL_MODE_NOT_FOUND')
226
227		cursor.execute(" SELECT name "
228		" FROM tbl_control_modes "
229		" WHERE name = %s AND id != %s ", (name, id_))
230		if cursor.fetchone() is not None:
231		cursor.close()
232		cnx.close()
233		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
234		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
235
236		# update control mode
237		update_row = (" UPDATE tbl_control_modes "
238		" SET name = %s, is_active = %s "
239		" WHERE id = %s ")
240		cursor.execute(update_row, (name,
241		is_active,
242		id_,))
243		cnx.commit()
244
245		cursor.close()
246		cnx.close()
247		resp.status = falcon.HTTP_200
248
249
250	View Code Duplication	class ControlModeExport:
		0 ignored issues – show Duplication introduced 2025-04-07 14:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
251		def __init__(self):
252		""""Initializes"""
253		pass
254
255		@staticmethod
256		def on_options(req, resp, id_):
257		resp.status = falcon.HTTP_200
258
259		@staticmethod
260		def on_get(req, resp, id_):
261		"""Handles GET requests"""
262		if 'API-KEY' not in req.headers or \
263		not isinstance(req.headers['API-KEY'], str) or \
264		len(str.strip(req.headers['API-KEY'])) == 0:
265		access_control(req)
266		else:
267		api_key_control(req)
268		if not id_.isdigit() or int(id_) <= 0:
269		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
270		description='API.INVALID_CONTROL_MODE_ID')
271
272		cnx = mysql.connector.connect(**config.myems_system_db)
273		cursor = cnx.cursor()
274
275		query = (" SELECT id, name, uuid, is_active "
276		" FROM tbl_control_modes "
277		" WHERE id = %s ")
278		cursor.execute(query, (id_,))
279		row = cursor.fetchone()
280		if row is None:
281		cursor.close()
282		cnx.close()
283		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
284		description='API.CONTROL_MODE_NOT_FOUND')
285
286		result = {"id": row[0],
287		"name": row[1],
288		"uuid": row[2],
289		"is_active": bool(row[3])}
290
291		cursor.close()
292		cnx.close()
293
294		resp.text = json.dumps(result)
295
296
297		class ControlModeImport:
298		def __init__(self):
299		""""Initializes"""
300		pass
301
302		@staticmethod
303		def on_options(req, resp):
304		resp.status = falcon.HTTP_200
305
306		@staticmethod
307		@user_logger
308		def on_post(req, resp):
309		"""Handles POST requests"""
310		admin_control(req)
311		try:
312		raw_json = req.stream.read().decode('utf-8')
313		except Exception as ex:
314		raise falcon.HTTPError(status=falcon.HTTP_400,
315		title='API.BAD_REQUEST',
316		description='API.FAILED_TO_READ_REQUEST_STREAM')
317		new_values = json.loads(raw_json)
318
319		if 'name' not in new_values.keys() or \
320		not isinstance(new_values['name'], str) or \
321		len(str.strip(new_values['name'])) == 0:
322		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
323		description='API.INVALID_METER_NAME')
324		name = str.strip(new_values['name'])
325
326		if 'is_active' not in new_values.keys() or \
327		not isinstance(new_values['is_active'], bool):
328		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
329		description='API.INVALID_IS_ACTIVE_VALUE')
330		is_active = new_values['is_active']
331
332		cnx = mysql.connector.connect(**config.myems_system_db)
333		cursor = cnx.cursor()
334
335		cursor.execute(" SELECT name "
336		" FROM tbl_control_modes "
337		" WHERE name = %s ", (name,))
338		if cursor.fetchone() is not None:
339		cursor.close()
340		cnx.close()
341		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
342		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
343
344		add_row = (" INSERT INTO tbl_control_modes "
345		" (name, uuid, is_active ) "
346		" VALUES (%s, %s, %s) ")
347		cursor.execute(add_row, (name,
348		str(uuid.uuid4()),
349		is_active,))
350		new_id = cursor.lastrowid
351		cnx.commit()
352
353		cursor.close()
354		cnx.close()
355
356		resp.status = falcon.HTTP_201
357		resp.location = '/controlmodes/' + str(new_id)
358
359
360	View Code Duplication	class ControlModeClone:
		0 ignored issues – show Duplication introduced 2025-04-07 14:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
361		def __init__(self):
362		""""Initializes"""
363		pass
364
365		@staticmethod
366		def on_options(req, resp, id_):
367		resp.status = falcon.HTTP_200
368
369		@staticmethod
370		@user_logger
371		def on_post(req, resp, id_):
372		if 'API-KEY' not in req.headers or \
373		not isinstance(req.headers['API-KEY'], str) or \
374		len(str.strip(req.headers['API-KEY'])) == 0:
375		access_control(req)
376		else:
377		api_key_control(req)
378		if not id_.isdigit() or int(id_) <= 0:
379		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
380		description='API.INVALID_CONTROL_MODE_ID')
381
382		cnx = mysql.connector.connect(**config.myems_system_db)
383		cursor = cnx.cursor()
384
385		query = (" SELECT id, name, uuid, is_active "
386		" FROM tbl_control_modes "
387		" WHERE id = %s ")
388		cursor.execute(query, (id_,))
389		row = cursor.fetchone()
390		if row is None:
391		cursor.close()
392		cnx.close()
393		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
394		description='API.CONTROL_MODE_NOT_FOUND')
395
396		result = {"id": row[0],
397		"name": row[1],
398		"uuid": row[2],
399		"is_active": bool(row[3])}
400
401		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
402		if config.utc_offset[0] == '-':
403		timezone_offset = -timezone_offset
404		new_name = (str.strip(result['name']) +
405		(datetime.utcnow() + timedelta(minutes=timezone_offset)).isoformat(sep='-', timespec='seconds'))
406		add_row = (" INSERT INTO tbl_control_modes "
407		" (name, uuid, is_active) "
408		" VALUES (%s, %s, %s) ")
409		cursor.execute(add_row, (new_name,
410		str(uuid.uuid4()),
411		result['is_active']))
412		new_id = cursor.lastrowid
413		cnx.commit()
414
415		cursor.close()
416		cnx.close()
417
418		resp.status = falcon.HTTP_201
419		resp.location = '/controlmodes/' + str(new_id)
420

MyEMS / myems

Push — master ( d2b0a0...60251a )

core.controlmode.ControlModeItem.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

core.controlmode.ControlModeItem.init() A