core.svg.SVGItem.on_delete() - Code Metrics - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

core.svg.SVGItem.on_delete() F
last analyzed 2025-08-16 07:04 UTC

↳ Parent: core.svg

Complexity

Conditions

Size

Total Lines	103
Code Lines	89

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	89
dl	0
loc	103
rs	1.2
c	0
b	0
f	0
cc	18
nop	3

How to fix Long Method Complexity

import uuid
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 SVGCollection:
    def __init__(self):
        """Initializes SVGCollection"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        # if turn quick mode on, do not return source code
        is_quick_mode = False
        if 'QUICKMODE' in req.headers and \
            isinstance(req.headers['QUICKMODE'], str) and \
                len(str.strip(req.headers['QUICKMODE'])) > 0 and \
                str.lower(req.headers['QUICKMODE']) in ('true', 't', 'on', 'yes', 'y'):
            is_quick_mode = True

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()
        if is_quick_mode:
            query = (" SELECT id, name, uuid, description "
                     " FROM tbl_svgs "
                     " ORDER BY id ")
            cursor.execute(query)
            rows_svgs = cursor.fetchall()

            result = list()
            if rows_svgs is not None and len(rows_svgs) > 0:
                for row in rows_svgs:

                    meta_result = {"id": row[0],
                                   "name": row[1],
                                   "uuid": row[2],
                                   "description": row[3]}
                    result.append(meta_result)
        else:
            query = (" SELECT id, name, uuid, source_code, description "
                     " FROM tbl_svgs "
                     " ORDER BY id ")
            cursor.execute(query)
            rows_svgs = cursor.fetchall()

            result = list()
            if rows_svgs is not None and len(rows_svgs) > 0:
                for row in rows_svgs:
                    meta_result = {"id": row[0],
                                   "name": row[1],
                                   "uuid": row[2],
                                   "source_code": row[3],
                                   "description": row[4]}
                    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:
            print(str(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_SVG_NAME')
        name = str.strip(new_values['data']['name'])

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

        if 'description' in new_values['data'].keys() and \
                new_values['data']['description'] is not None and \
                len(str(new_values['data']['description'])) > 0:
            description = str.strip(new_values['data']['description'])
        else:
            description = None

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

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

        add_values = (" INSERT INTO tbl_svgs "
                      "    (name, uuid, source_code, description) "
                      " VALUES (%s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    source_code,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


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

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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SVG_ID')

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

        query = (" SELECT id, name, uuid, source_code, description "
                 " FROM tbl_svgs "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()
        cursor.close()
        cnx.close()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.SVG_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "source_code": row[3],
                           "description": row[4]}

        resp.text = json.dumps(meta_result)

    @staticmethod
    @user_logger
    def on_delete(req, resp, id_):
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SVG_ID')
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        cursor.execute(" SELECT name "
                       " FROM tbl_svgs "
                       " 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.SVG_NOT_FOUND')
        # check if any equipment is bound to this SVG
        cursor.execute("SELECT id FROM tbl_equipments WHERE svg_id = %s", (id_,))
        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_EQUIPMENTS')
        # check if any combined equipment is bound to this SVG
        cursor.execute("SELECT id FROM tbl_combined_equipments WHERE svg_id = %s", (id_,))
        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_COMBINED_EQUIPMENTS')
        # check if any distribution system is bound to this SVG
        cursor.execute("SELECT id FROM tbl_distribution_systems WHERE svg_id = %s", (id_,))
        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_DISTRIBUTION_SYSTEMS')
        # check if any energy storage power station is bound to this SVG
        for col in ['svg_id', 'svg2_id', 'svg3_id', 'svg4_id', 'svg5_id']:
            cursor.execute(f"SELECT id FROM tbl_energy_storage_power_stations WHERE {col} = %s", (id_,))
            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_ENERGY_STORAGE_POWER_STATIONS')
        # check if any photovoltaic power station is bound to this SVG
        for col in ['svg_id', 'svg2_id', 'svg3_id', 'svg4_id', 'svg5_id']:
            cursor.execute(f"SELECT id FROM tbl_photovoltaic_power_stations WHERE {col} = %s", (id_,))
            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_PHOTOVOLTAIC_POWER_STATIONS')
        # check if any microgrid is bound to this SVG
        for col in ['svg_id', 'svg2_id', 'svg3_id', 'svg4_id', 'svg5_id']:
            cursor.execute(f"SELECT id FROM tbl_microgrids WHERE {col} = %s", (id_,))
            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_MICROGRIDS')
        # check if any virtual power plant is bound to this SVG
        cursor.execute("SELECT id FROM tbl_virtual_power_plants WHERE svg_id = %s", (id_,))
        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_VIRTUAL_POWER_PLANTS')
        # check if any wind farm is bound to this SVG
        for col in ['svg_id', 'svg2_id', 'svg3_id', 'svg4_id', 'svg5_id']:
            cursor.execute(f"SELECT id FROM tbl_wind_farms WHERE {col} = %s", (id_,))
            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_WIND_FARMS')
        # check if any charging station is bound to this SVG
        for col in ['svg_id', 'svg2_id', 'svg3_id', 'svg4_id', 'svg5_id']:
            cursor.execute(f"SELECT id FROM tbl_charging_stations WHERE {col} = %s", (id_,))
            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_CHARGING_STATIONS')

        cursor.execute(" DELETE FROM tbl_svgs 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:
            print(str(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_SVG_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_SVG_NAME')
        name = str.strip(new_values['data']['name'])

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

        if 'description' in new_values['data'].keys() and \
                new_values['data']['description'] is not None and \
                len(str(new_values['data']['description'])) > 0:
            description = str.strip(new_values['data']['description'])
        else:
            description = None

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

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

        update_row = (" UPDATE tbl_svgs "
                      " SET name = %s, source_code = %s, description = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (name,
                                    source_code,
                                    description,
                                    id_))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200


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

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

    @staticmethod
    def on_get(req, resp, id_):
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SVG_ID')

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

        query = (" SELECT id, name, uuid, source_code, description "
                 " FROM tbl_svgs "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.SVG_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "source_code": row[3],
                           "description": row[4]}
        cursor.close()
        cnx.close()
        resp.text = json.dumps(meta_result)


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

    @staticmethod
    def on_options(req, resp):
        _ = req
        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:
            print(str(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_SVG_NAME')
        name = str.strip(new_values['name'])

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

        if 'description' in new_values.keys() and \
                new_values['description'] is not None and \
                len(str(new_values['description'])) > 0:
            description = str.strip(new_values['description'])
        else:
            description = None

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

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

        add_values = (" INSERT INTO tbl_svgs "
                      "    (name, uuid, source_code, description) "
                      " VALUES (%s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    source_code,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


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

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

    @staticmethod
    @user_logger
    def on_post(req, resp, id_):
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SVG_ID')

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

        query = (" SELECT id, name, uuid, source_code, description "
                 " FROM tbl_svgs "
                 " WHERE id = %s ")
        cursor.execute(query, (id_,))
        row = cursor.fetchone()

        if row is None:
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.SVG_NOT_FOUND')
        else:
            meta_result = {"id": row[0],
                           "name": row[1],
                           "uuid": row[2],
                           "source_code": row[3],
                           "description": row[4]}

            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(meta_result['name']) +
                        (datetime.utcnow() + timedelta(minutes=timezone_offset)).isoformat(sep='-', timespec='seconds'))
            add_values = (" INSERT INTO tbl_svgs "
                          "    (name, uuid, source_code, description) "
                          " VALUES (%s, %s, %s, %s) ")
            cursor.execute(add_values, (new_name,
                                        str(uuid.uuid4()),
                                        meta_result['source_code'],
                                        meta_result['description']))
            new_id = cursor.lastrowid
            cnx.commit()
            cursor.close()
            cnx.close()

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


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

MyEMS / myems

core.svg.SVGItem.on_delete() F last analyzed 2025-08-16 07:04 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

core.svg.SVGItem.on_delete() F
last analyzed 2025-08-16 07:04 UTC