core.controlmode.ControlModeCollection.on_post() - Code Metrics - Inspection of "added control modes and control times" - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 60251a...b3cd43 )

unknown

created 2025-04-08 02:33 UTC

core.controlmode.ControlModeCollection.on_post() C

↳ Parent: core.controlmode

Complexity

Conditions

Size

Total Lines	66
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	52
dl	0
loc	66
rs	5.7709
c	0
b	0
f	0
cc	10
nop	2

How to fix Long Method Complexity

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]), 'times': list()}

                query = (" SELECT start_time_of_day, end_time_of_day, power_value "
                         " FROM tbl_control_modes_times "
                         " WHERE control_mode_id = %s  "
                         " ORDER BY id")
                cursor.execute(query, (meta_result['id'],))
                rows_times = cursor.fetchall()
                if rows_times is not None and len(rows_times) > 0:
                    for row_time in rows_times:
                        meta_data = {"start_time_of_day": str(row_time[0]),
                                     "end_time_of_day": str(row_time[1]),
                                     "power_value": row_time[2]}
                        meta_result['times'].append(meta_data)
                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_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']

        if new_values['data']['times'] is None:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_CONTROL_MODE_TIMES')

        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()
        for item in new_values['data']['times']:
            add_time = (" INSERT INTO tbl_control_modes_times "
                        "             (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
                        " VALUES (%s, %s, %s, %s) ")
            cursor.execute(add_time, (new_id,
                                      item['start_time_of_day'],
                                      item['end_time_of_day'],
                                      item['power_value']))
            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]), 'times': list()}

        query = (" SELECT start_time_of_day, end_time_of_day, power_value "
                 " FROM tbl_control_modes_times "
                 " WHERE control_mode_id = %s  "
                 " ORDER BY id ")
        cursor.execute(query, (result['id'],))
        rows_times = cursor.fetchall()
        if rows_times is not None and len(rows_times) > 0:
            for row_time in rows_times:
                meta_data = {"start_time_of_day": str(row_time[0]),
                             "end_time_of_day": str(row_time[1]),
                             "power_value": row_time[2]}
                result['times'].append(meta_data)

        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_times WHERE control_mode_id = %s ", (id_,))
        cnx.commit()

        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']

        if 'times' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['times'], list) or \
                len(new_values['data']['times']) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.INVALID_CONTROL_MODE_TIMES')

        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()

        # remove all (possible) exist times
        cursor.execute(" DELETE FROM tbl_control_modes_times "
                       " WHERE control_mode_id = %s ",
                       (id_,))
        cnx.commit()

        for item in new_values['data']['times']:
            add_time = (" INSERT INTO tbl_control_modes_times "
                        "             (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
                        " VALUES (%s, %s, %s, %s) ")
            cursor.execute(add_time, (id_,
                                      item['start_time_of_day'],
                                      item['end_time_of_day'],
                                      item['power_value']))
            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]), 'times': list()}

        query = (" SELECT start_time_of_day, end_time_of_day, power_value "
                 " FROM tbl_control_modes_times "
                 " WHERE control_mode_id = %s  "
                 " ORDER BY id")
        cursor.execute(query, (result['id'],))
        rows_times = cursor.fetchall()
        if rows_times is not None and len(rows_times) > 0:
            for row_time in rows_times:
                meta_data = {"start_time_of_day": str(row_time[0]),
                             "end_time_of_day": str(row_time[1]),
                             "power_value": row_time[2]}
                result['times'].append(meta_data)

        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']

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

        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()

        for item in new_values['times']:
            add_time = (" INSERT INTO tbl_control_modes_times "
                        "             (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
                        " VALUES (%s, %s, %s, %s) ")
            cursor.execute(add_time, (new_id,
                                      item['start_time_of_day'],
                                      item['end_time_of_day'],
                                      item['power_value']))
            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]), 'times': list()}

        query = (" SELECT start_time_of_day, end_time_of_day, power_value "
                 " FROM tbl_control_modes_times "
                 " WHERE control_mode_id = %s  "
                 " ORDER BY id")
        cursor.execute(query, (result['id'],))
        rows_times = cursor.fetchall()
        if rows_times is not None and len(rows_times) > 0:
            for row_time in rows_times:
                meta_data = {"start_time_of_day": str(row_time[0]),
                             "end_time_of_day": str(row_time[1]),
                             "power_value": row_time[2]}
                result['times'].append(meta_data)

        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()
        for item in result['times']:
            add_time = (" INSERT INTO tbl_control_modes_times "
                        "             (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
                        " VALUES (%s, %s, %s, %s) ")
            cursor.execute(add_time, (new_id,
                                      item['start_time_of_day'],
                                      item['end_time_of_day'],
                                      item['power_value']))
            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], "name": row[1], "uuid": row[2], "is_active": bool(row[3]), 'times': list()}
40
41		query = (" SELECT start_time_of_day, end_time_of_day, power_value "
42		" FROM tbl_control_modes_times "
43		" WHERE control_mode_id = %s "
44		" ORDER BY id")
45		cursor.execute(query, (meta_result['id'],))
46		rows_times = cursor.fetchall()
47		if rows_times is not None and len(rows_times) > 0:
48		for row_time in rows_times:
49		meta_data = {"start_time_of_day": str(row_time[0]),
50		"end_time_of_day": str(row_time[1]),
51		"power_value": row_time[2]}
52		meta_result['times'].append(meta_data)
53		result.append(meta_result)
54
55		cursor.close()
56		cnx.close()
57
58		resp.text = json.dumps(result)
59
60		@staticmethod
61		@user_logger
62		def on_post(req, resp):
63		"""Handles POST requests"""
64		admin_control(req)
65		try:
66		raw_json = req.stream.read().decode('utf-8')
67		except Exception as ex:
68		raise falcon.HTTPError(status=falcon.HTTP_400,
69		title='API.BAD_REQUEST',
70		description='API.FAILED_TO_READ_REQUEST_STREAM')
71		new_values = json.loads(raw_json)
72
73		if 'name' not in new_values['data'].keys() or \
74		not isinstance(new_values['data']['name'], str) or \
75		len(str.strip(new_values['data']['name'])) == 0:
76		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
77		description='API.INVALID_CONTROL_MODE_NAME')
78		name = str.strip(new_values['data']['name'])
79
80		if 'is_active' not in new_values['data'].keys() or \
81		not isinstance(new_values['data']['is_active'], bool):
82		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
83		description='API.INVALID_IS_ACTIVE_VALUE')
84		is_active = new_values['data']['is_active']
85
86		if new_values['data']['times'] is None:
87		raise falcon.HTTPError(status=falcon.HTTP_400,
88		title='API.BAD_REQUEST',
89		description='API.INVALID_CONTROL_MODE_TIMES')
90
91		cnx = mysql.connector.connect(**config.myems_system_db)
92		cursor = cnx.cursor()
93
94		cursor.execute(" SELECT name "
95		" FROM tbl_control_modes "
96		" WHERE name = %s ", (name,))
97		if cursor.fetchone() is not None:
98		cursor.close()
99		cnx.close()
100		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
101		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
102
103		add_row = (" INSERT INTO tbl_control_modes "
104		" (name, uuid, is_active ) "
105		" VALUES (%s, %s, %s) ")
106		cursor.execute(add_row, (name,
107		str(uuid.uuid4()),
108		is_active))
109		new_id = cursor.lastrowid
110		cnx.commit()
111		for item in new_values['data']['times']:
112		add_time = (" INSERT INTO tbl_control_modes_times "
113		" (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
114		" VALUES (%s, %s, %s, %s) ")
115		cursor.execute(add_time, (new_id,
116		item['start_time_of_day'],
117		item['end_time_of_day'],
118		item['power_value']))
119		cnx.commit()
120
121		cursor.close()
122		cnx.close()
123
124		resp.status = falcon.HTTP_201
125		resp.location = '/controlmodes/' + str(new_id)
126
127
128		class ControlModeItem:
129		def __init__(self):
130		""""Initializes"""
131		pass
132
133		@staticmethod
134		def on_options(req, resp, id_):
135		resp.status = falcon.HTTP_200
136
137		@staticmethod
138		def on_get(req, resp, id_):
139		"""Handles GET requests"""
140		if 'API-KEY' not in req.headers or \
141		not isinstance(req.headers['API-KEY'], str) or \
142		len(str.strip(req.headers['API-KEY'])) == 0:
143		access_control(req)
144		else:
145		api_key_control(req)
146		if not id_.isdigit() or int(id_) <= 0:
147		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
148		description='API.INVALID_CONTROL_MODE_ID')
149
150		cnx = mysql.connector.connect(**config.myems_system_db)
151		cursor = cnx.cursor()
152
153		query = (" SELECT id, name, uuid, is_active "
154		" FROM tbl_control_modes "
155		" WHERE id = %s ")
156		cursor.execute(query, (id_,))
157		row = cursor.fetchone()
158		if row is None:
159		cursor.close()
160		cnx.close()
161		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
162		description='API.CONTROL_MODE_NOT_FOUND')
163
164		result = {"id": row[0], "name": row[1], "uuid": row[2], "is_active": bool(row[3]), 'times': list()}
165
166		query = (" SELECT start_time_of_day, end_time_of_day, power_value "
167		" FROM tbl_control_modes_times "
168		" WHERE control_mode_id = %s "
169		" ORDER BY id ")
170		cursor.execute(query, (result['id'],))
171		rows_times = cursor.fetchall()
172		if rows_times is not None and len(rows_times) > 0:
173		for row_time in rows_times:
174		meta_data = {"start_time_of_day": str(row_time[0]),
175		"end_time_of_day": str(row_time[1]),
176		"power_value": row_time[2]}
177		result['times'].append(meta_data)
178
179		cursor.close()
180		cnx.close()
181
182		resp.text = json.dumps(result)
183
184		@staticmethod
185		@user_logger
186		def on_delete(req, resp, id_):
187		"""Handles DELETE requests"""
188		admin_control(req)
189		if not id_.isdigit() or int(id_) <= 0:
190		raise falcon.HTTPError(status=falcon.HTTP_400,
191		title='API.BAD_REQUEST',
192		description='API.INVALID_CONTROL_MODE_ID')
193
194		cnx = mysql.connector.connect(**config.myems_system_db)
195		cursor = cnx.cursor()
196
197		cursor.execute(" SELECT name "
198		" FROM tbl_control_modes "
199		" WHERE id = %s ", (id_,))
200		if cursor.fetchone() is None:
201		cursor.close()
202		cnx.close()
203		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
204		description='API.CONTROL_MODE_NOT_FOUND')
205
206		cursor.execute(" DELETE FROM tbl_control_modes_times WHERE control_mode_id = %s ", (id_,))
207		cnx.commit()
208
209		cursor.execute(" DELETE FROM tbl_control_modes WHERE id = %s ", (id_,))
210		cnx.commit()
211
212		cursor.close()
213		cnx.close()
214
215		resp.status = falcon.HTTP_204
216
217		@staticmethod
218		@user_logger
219		def on_put(req, resp, id_):
220		"""Handles PUT requests"""
221		admin_control(req)
222		try:
223		raw_json = req.stream.read().decode('utf-8')
224		except Exception as ex:
225		raise falcon.HTTPError(status=falcon.HTTP_400,
226		title='API.BAD_REQUEST',
227		description='API.FAILED_TO_READ_REQUEST_STREAM')
228
229		if not id_.isdigit() or int(id_) <= 0:
230		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
231		description='API.INVALID_CONTROL_MODE_ID')
232
233		new_values = json.loads(raw_json)
234
235		if 'name' not in new_values['data'].keys() or \
236		not isinstance(new_values['data']['name'], str) or \
237		len(str.strip(new_values['data']['name'])) == 0:
238		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
239		description='API.INVALID_CONTROL_MODE_NAME')
240		name = str.strip(new_values['data']['name'])
241
242		if 'is_active' not in new_values['data'].keys() or \
243		not isinstance(new_values['data']['is_active'], bool):
244		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
245		description='API.INVALID_IS_ACTIVE_VALUE')
246		is_active = new_values['data']['is_active']
247
248		if 'times' not in new_values['data'].keys() or \
249		not isinstance(new_values['data']['times'], list) or \
250		len(new_values['data']['times']) == 0:
251		raise falcon.HTTPError(status=falcon.HTTP_400,
252		title='API.BAD_REQUEST',
253		description='API.INVALID_CONTROL_MODE_TIMES')
254
255		cnx = mysql.connector.connect(**config.myems_system_db)
256		cursor = cnx.cursor()
257
258		# check if the control mode exists
259		query = (" SELECT name "
260		" FROM tbl_control_modes "
261		" WHERE id = %s ")
262		cursor.execute(query, (id_,))
263		cursor.fetchone()
264
265		if cursor.rowcount != 1:
266		cursor.close()
267		cnx.close()
268		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
269		description='API.CONTROL_MODE_NOT_FOUND')
270
271		cursor.execute(" SELECT name "
272		" FROM tbl_control_modes "
273		" WHERE name = %s AND id != %s ", (name, id_))
274		if cursor.fetchone() is not None:
275		cursor.close()
276		cnx.close()
277		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
278		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
279
280		# update control mode
281		update_row = (" UPDATE tbl_control_modes "
282		" SET name = %s, is_active = %s "
283		" WHERE id = %s ")
284		cursor.execute(update_row, (name,
285		is_active,
286		id_,))
287		cnx.commit()
288
289		# remove all (possible) exist times
290		cursor.execute(" DELETE FROM tbl_control_modes_times "
291		" WHERE control_mode_id = %s ",
292		(id_,))
293		cnx.commit()
294
295		for item in new_values['data']['times']:
296		add_time = (" INSERT INTO tbl_control_modes_times "
297		" (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
298		" VALUES (%s, %s, %s, %s) ")
299		cursor.execute(add_time, (id_,
300		item['start_time_of_day'],
301		item['end_time_of_day'],
302		item['power_value']))
303		cnx.commit()
304		cursor.close()
305		cnx.close()
306		resp.status = falcon.HTTP_200
307
308
309	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...
310		def __init__(self):
311		""""Initializes"""
312		pass
313
314		@staticmethod
315		def on_options(req, resp, id_):
316		resp.status = falcon.HTTP_200
317
318		@staticmethod
319		def on_get(req, resp, id_):
320		"""Handles GET requests"""
321		if 'API-KEY' not in req.headers or \
322		not isinstance(req.headers['API-KEY'], str) or \
323		len(str.strip(req.headers['API-KEY'])) == 0:
324		access_control(req)
325		else:
326		api_key_control(req)
327		if not id_.isdigit() or int(id_) <= 0:
328		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
329		description='API.INVALID_CONTROL_MODE_ID')
330
331		cnx = mysql.connector.connect(**config.myems_system_db)
332		cursor = cnx.cursor()
333
334		query = (" SELECT id, name, uuid, is_active "
335		" FROM tbl_control_modes "
336		" WHERE id = %s ")
337		cursor.execute(query, (id_,))
338		row = cursor.fetchone()
339		if row is None:
340		cursor.close()
341		cnx.close()
342		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
343		description='API.CONTROL_MODE_NOT_FOUND')
344
345		result = {"id": row[0], "name": row[1], "uuid": row[2], "is_active": bool(row[3]), 'times': list()}
346
347		query = (" SELECT start_time_of_day, end_time_of_day, power_value "
348		" FROM tbl_control_modes_times "
349		" WHERE control_mode_id = %s "
350		" ORDER BY id")
351		cursor.execute(query, (result['id'],))
352		rows_times = cursor.fetchall()
353		if rows_times is not None and len(rows_times) > 0:
354		for row_time in rows_times:
355		meta_data = {"start_time_of_day": str(row_time[0]),
356		"end_time_of_day": str(row_time[1]),
357		"power_value": row_time[2]}
358		result['times'].append(meta_data)
359
360		cursor.close()
361		cnx.close()
362
363		resp.text = json.dumps(result)
364
365
366		class ControlModeImport:
367		def __init__(self):
368		""""Initializes"""
369		pass
370
371		@staticmethod
372		def on_options(req, resp):
373		resp.status = falcon.HTTP_200
374
375		@staticmethod
376		@user_logger
377		def on_post(req, resp):
378		"""Handles POST requests"""
379		admin_control(req)
380		try:
381		raw_json = req.stream.read().decode('utf-8')
382		except Exception as ex:
383		raise falcon.HTTPError(status=falcon.HTTP_400,
384		title='API.BAD_REQUEST',
385		description='API.FAILED_TO_READ_REQUEST_STREAM')
386		new_values = json.loads(raw_json)
387
388		if 'name' not in new_values.keys() or \
389		not isinstance(new_values['name'], str) or \
390		len(str.strip(new_values['name'])) == 0:
391		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
392		description='API.INVALID_METER_NAME')
393		name = str.strip(new_values['name'])
394
395		if 'is_active' not in new_values.keys() or \
396		not isinstance(new_values['is_active'], bool):
397		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
398		description='API.INVALID_IS_ACTIVE_VALUE')
399		is_active = new_values['is_active']
400
401		if 'times' not in new_values.keys() or \
402		not isinstance(new_values['times'], list) or \
403		len(new_values['times']) == 0:
404		raise falcon.HTTPError(status=falcon.HTTP_400,
405		title='API.BAD_REQUEST',
406		description='API.INVALID_CONTROL_MODE_TIMES')
407
408		cnx = mysql.connector.connect(**config.myems_system_db)
409		cursor = cnx.cursor()
410
411		cursor.execute(" SELECT name "
412		" FROM tbl_control_modes "
413		" WHERE name = %s ", (name,))
414		if cursor.fetchone() is not None:
415		cursor.close()
416		cnx.close()
417		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
418		description='API.CONTROL_MODE_NAME_IS_ALREADY_IN_USE')
419
420		add_row = (" INSERT INTO tbl_control_modes "
421		" (name, uuid, is_active ) "
422		" VALUES (%s, %s, %s) ")
423		cursor.execute(add_row, (name,
424		str(uuid.uuid4()),
425		is_active,))
426		new_id = cursor.lastrowid
427		cnx.commit()
428
429		for item in new_values['times']:
430		add_time = (" INSERT INTO tbl_control_modes_times "
431		" (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
432		" VALUES (%s, %s, %s, %s) ")
433		cursor.execute(add_time, (new_id,
434		item['start_time_of_day'],
435		item['end_time_of_day'],
436		item['power_value']))
437		cnx.commit()
438
439		cursor.close()
440		cnx.close()
441
442		resp.status = falcon.HTTP_201
443		resp.location = '/controlmodes/' + str(new_id)
444
445
446		class ControlModeClone:
447		def __init__(self):
448		""""Initializes"""
449		pass
450
451		@staticmethod
452		def on_options(req, resp, id_):
453		resp.status = falcon.HTTP_200
454
455		@staticmethod
456		@user_logger
457		def on_post(req, resp, id_):
458		if 'API-KEY' not in req.headers or \
459		not isinstance(req.headers['API-KEY'], str) or \
460		len(str.strip(req.headers['API-KEY'])) == 0:
461		access_control(req)
462		else:
463		api_key_control(req)
464		if not id_.isdigit() or int(id_) <= 0:
465		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
466		description='API.INVALID_CONTROL_MODE_ID')
467
468		cnx = mysql.connector.connect(**config.myems_system_db)
469		cursor = cnx.cursor()
470
471		query = (" SELECT id, name, uuid, is_active "
472		" FROM tbl_control_modes "
473		" WHERE id = %s ")
474		cursor.execute(query, (id_,))
475		row = cursor.fetchone()
476		if row is None:
477		cursor.close()
478		cnx.close()
479		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
480		description='API.CONTROL_MODE_NOT_FOUND')
481
482		result = {"id": row[0], "name": row[1], "uuid": row[2], "is_active": bool(row[3]), 'times': list()}
483
484		query = (" SELECT start_time_of_day, end_time_of_day, power_value "
485		" FROM tbl_control_modes_times "
486		" WHERE control_mode_id = %s "
487		" ORDER BY id")
488		cursor.execute(query, (result['id'],))
489		rows_times = cursor.fetchall()
490		if rows_times is not None and len(rows_times) > 0:
491		for row_time in rows_times:
492		meta_data = {"start_time_of_day": str(row_time[0]),
493		"end_time_of_day": str(row_time[1]),
494		"power_value": row_time[2]}
495		result['times'].append(meta_data)
496
497		timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
498		if config.utc_offset[0] == '-':
499		timezone_offset = -timezone_offset
500		new_name = (str.strip(result['name']) +
501		(datetime.utcnow() + timedelta(minutes=timezone_offset)).isoformat(sep='-', timespec='seconds'))
502		add_row = (" INSERT INTO tbl_control_modes "
503		" (name, uuid, is_active) "
504		" VALUES (%s, %s, %s) ")
505		cursor.execute(add_row, (new_name,
506		str(uuid.uuid4()),
507		result['is_active']))
508		new_id = cursor.lastrowid
509		cnx.commit()
510		for item in result['times']:
511		add_time = (" INSERT INTO tbl_control_modes_times "
512		" (control_mode_id, start_time_of_day, end_time_of_day, power_value) "
513		" VALUES (%s, %s, %s, %s) ")
514		cursor.execute(add_time, (new_id,
515		item['start_time_of_day'],
516		item['end_time_of_day'],
517		item['power_value']))
518		cnx.commit()
519		cursor.close()
520		cnx.close()
521
522		resp.status = falcon.HTTP_201
523		resp.location = '/controlmodes/' + str(new_id)
524

MyEMS / myems

Push — master ( 60251a...b3cd43 )

core.controlmode.ControlModeCollection.on_post() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like