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

core.emailserver.EmailServerItem.on_put() F
last analyzed 2025-12-05 00:16 UTC

↳ Parent: core.emailserver

Complexity

Conditions

Size

Total Lines	111
Code Lines	87

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	87
dl	0
loc	111
rs	0
c	0
b	0
f	0
cc	26
nop	3

How to fix Long Method Complexity

import base64
import re
import falcon
import mysql.connector
import simplejson as json
from core.useractivity import user_logger, admin_control
import config


class EmailServerCollection:
    """
    Email Server Collection Resource

    This class handles CRUD operations for email server collection.
    It provides endpoints for listing all email servers and creating new servers.
    Email servers configure SMTP settings for sending notifications.
    """
    def __init__(self):
        """Initialize EmailServerCollection"""
        pass

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

    @staticmethod
    def on_get(req, resp):
        admin_control(req)
        cnx = mysql.connector.connect(**config.myems_fdd_db)
        cursor = cnx.cursor()

        query = (" SELECT id, host, port, requires_authentication, user_name, password, from_addr "
                 " FROM tbl_email_servers ")
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        cnx.close()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0],
                               "host": row[1],
                               "port": row[2],
                               "requires_authentication": bool(row[3]),
                               "user_name": row[4],
                               "password": str(base64.b64decode(bytearray(row[5], 'utf-8')), 'utf-8')
                               if row[5] is not None else None,
                               "from_addr": row[6]}
                result.append(meta_result)

        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(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 'host' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['host'], str) or \
                len(str.strip(new_values['data']['host'])) == 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_EMAIL_SERVER_HOST')

        host = str.strip(new_values['data']['host'])

        if 'port' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['port'], int) or \
                new_values['data']['port'] <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PORT')
        port = float(new_values['data']['port'])

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

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

        if requires_authentication:
            if 'password' not in new_values['data'].keys() or \
                    not isinstance(new_values['data']['password'], str) or \
                    len(str.strip(new_values['data']['password'])) == 0:
                raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.INVALID_PASSWORD')
            password = base64.b64encode(bytearray(new_values['data']['password'], 'utf-8'))
        else:
            password = None

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

        match = re.match(r'^[_A-Za-z0-9-]+(\.[_A-Za-z0-9-]+)*@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)*(\.[A-Za-z]{2,4})$',
                         from_addr)
        if match is None:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_FROM_ADDR')

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

        cursor.execute(" SELECT host "
                       " FROM tbl_email_servers "
                       " WHERE host = %s ", (host,))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.BAD_REQUEST',
                                   description='API.EMAIL_SERVER_HOST_IS_ALREADY_IN_USE')

        add_value = (" INSERT INTO tbl_email_servers "
                     "    (host, port, requires_authentication, user_name, password, from_addr) "
                     " VALUES (%s, %s, %s, %s, %s, %s) ")
        cursor.execute(add_value, (host,
                                   port,
                                   requires_authentication,
                                   user_name,
                                   password,
                                   from_addr))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class EmailServerItem:
    def __init__(self):
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        admin_control(req)
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='400 Bad Request')

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

        query = (" SELECT id, host, port, requires_authentication, user_name, password, from_addr "
                 " FROM tbl_email_servers "
                 " 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')

        result = {"id": row[0],
                  "host": row[1],
                  "port": row[2],
                  "requires_authentication": bool(row[3]),
                  "user_name": row[4],
                  "password": str(base64.b64decode(bytearray(row[5], 'utf-8')), 'utf-8')
                  if row[5] is not None else None,
                  "from_addr": row[5]}
        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_EMAIL_SERVER_ID')

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

        cursor.execute(" SELECT host "
                       " FROM tbl_email_servers "
                       " 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.EMAIL_SERVER_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_email_servers 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_EMAIL_SERVER_ID')

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

        host = str.strip(new_values['data']['host'])

        if 'port' not in new_values['data'].keys() or \
                not isinstance(new_values['data']['port'], int) or \
                new_values['data']['port'] <= 0:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_PORT')
        port = float(new_values['data']['port'])

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

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

        if requires_authentication:
            if 'password' not in new_values['data'].keys() or \
                    not isinstance(new_values['data']['password'], str) or \
                    len(str.strip(new_values['data']['password'])) == 0:
                raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.INVALID_PASSWORD')
            password = base64.b64encode(bytearray(new_values['data']['password'], 'utf-8'))
        else:
            password = None

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

        match = re.match(r'^[_A-Za-z0-9-]+(\.[_A-Za-z0-9-]+)*@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)*(\.[A-Za-z]{2,4})$',
                         from_addr)
        if match is None:
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_FROM_ADDR')

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

        cursor.execute(" SELECT id "
                       " FROM tbl_email_servers "
                       " 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.EMAIL_SERVER_NOT_FOUND')

        cursor.execute(" SELECT host "
                       " FROM tbl_email_servers "
                       " WHERE host = %s AND id != %s ", (host, id_))
        if cursor.fetchone() is not None:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_404, title='API.BAD_REQUEST',
                                   description='API.EMAIL_SERVER_HOST_IS_ALREADY_IN_USE')

        update_row = (" UPDATE tbl_email_servers "
                      " SET host = %s, port = %s, requires_authentication = %s, "
                      "     user_name = %s, password = %s, from_addr = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (host,
                                    port,
                                    requires_authentication,
                                    user_name,
                                    password,
                                    from_addr,
                                    id_,))
        cnx.commit()

        cursor.close()
        cnx.close()

        resp.status = falcon.HTTP_200


1			import base64
2			import re
3			import falcon
4			import mysql.connector
5			import simplejson as json
6			from core.useractivity import user_logger, admin_control
7			import config
8
9
10			class EmailServerCollection:
11			"""
12			Email Server Collection Resource
13
14			This class handles CRUD operations for email server collection.
15			It provides endpoints for listing all email servers and creating new servers.
16			Email servers configure SMTP settings for sending notifications.
17			"""
18			def __init__(self):
19			"""Initialize EmailServerCollection"""
20			pass
21
22			@staticmethod
23			def on_options(req, resp):
24			"""Handle OPTIONS requests for CORS preflight"""
25			_ = req
26			resp.status = falcon.HTTP_200
27
28			@staticmethod
29			def on_get(req, resp):
30			admin_control(req)
31			cnx = mysql.connector.connect(**config.myems_fdd_db)
32			cursor = cnx.cursor()
33
34			query = (" SELECT id, host, port, requires_authentication, user_name, password, from_addr "
35			" FROM tbl_email_servers ")
36			cursor.execute(query)
37			rows = cursor.fetchall()
38			cursor.close()
39			cnx.close()
40
41			result = list()
42			if rows is not None and len(rows) > 0:
43			for row in rows:
44			meta_result = {"id": row[0],
45			"host": row[1],
46			"port": row[2],
47			"requires_authentication": bool(row[3]),
48			"user_name": row[4],
49			"password": str(base64.b64decode(bytearray(row[5], 'utf-8')), 'utf-8')
50			if row[5] is not None else None,
51			"from_addr": row[6]}
52			result.append(meta_result)
53
54			resp.text = json.dumps(result)
55
56			@staticmethod
57			@user_logger
58			def on_post(req, resp):
59			"""Handles POST requests"""
60			admin_control(req)
61			try:
62			raw_json = req.stream.read().decode('utf-8')
63			except Exception as ex:
64			print(ex)
65			raise falcon.HTTPError(status=falcon.HTTP_400,
66			title='API.BAD_REQUEST',
67			description='API.FAILED_TO_READ_REQUEST_STREAM')
68
69			new_values = json.loads(raw_json)
70
71			if 'host' not in new_values['data'].keys() or \
72			not isinstance(new_values['data']['host'], str) or \
73			len(str.strip(new_values['data']['host'])) == 0:
74			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
75			description='API.INVALID_EMAIL_SERVER_HOST')
76
77			host = str.strip(new_values['data']['host'])
78
79			if 'port' not in new_values['data'].keys() or \
80			not isinstance(new_values['data']['port'], int) or \
81			new_values['data']['port'] <= 0:
82			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
83			description='API.INVALID_PORT')
84			port = float(new_values['data']['port'])
85
86			if 'requires_authentication' not in new_values['data'].keys() or \
87			not isinstance(new_values['data']['requires_authentication'], bool):
88			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
89			description='API.INVALID_REQUIRES_AUTHENTICATION')
90			requires_authentication = new_values['data']['requires_authentication']
91
92			if requires_authentication:
93			if 'user_name' not in new_values['data'].keys() or \
94			not isinstance(new_values['data']['user_name'], str) or \
95			len(str.strip(new_values['data']['user_name'])) == 0:
96			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
97			description='API.INVALID_USER_NAME')
98			user_name = new_values['data']['user_name']
99			else:
100			user_name = None
101
102			if requires_authentication:
103			if 'password' not in new_values['data'].keys() or \
104			not isinstance(new_values['data']['password'], str) or \
105			len(str.strip(new_values['data']['password'])) == 0:
106			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
107			description='API.INVALID_PASSWORD')
108			password = base64.b64encode(bytearray(new_values['data']['password'], 'utf-8'))
109			else:
110			password = None
111
112			if 'from_addr' not in new_values['data'].keys() or \
113			not isinstance(new_values['data']['from_addr'], str) or \
114			len(str.strip(new_values['data']['from_addr'])) == 0:
115			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
116			description='API.INVALID_FROM_ADDR')
117			from_addr = new_values['data']['from_addr']
118
119			match = re.match(r'^[_A-Za-z0-9-]+(\.[_A-Za-z0-9-]+)@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)(\.[A-Za-z]{2,4})$',
120			from_addr)
121			if match is None:
122			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
123			description='API.INVALID_FROM_ADDR')
124
125			cnx = mysql.connector.connect(**config.myems_fdd_db)
126			cursor = cnx.cursor()
127
128			cursor.execute(" SELECT host "
129			" FROM tbl_email_servers "
130			" WHERE host = %s ", (host,))
131			if cursor.fetchone() is not None:
132			cursor.close()
133			cnx.close()
134			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.BAD_REQUEST',
135			description='API.EMAIL_SERVER_HOST_IS_ALREADY_IN_USE')
136
137			add_value = (" INSERT INTO tbl_email_servers "
138			" (host, port, requires_authentication, user_name, password, from_addr) "
139			" VALUES (%s, %s, %s, %s, %s, %s) ")
140			cursor.execute(add_value, (host,
141			port,
142			requires_authentication,
143			user_name,
144			password,
145			from_addr))
146			new_id = cursor.lastrowid
147			cnx.commit()
148			cursor.close()
149			cnx.close()
150
151			resp.status = falcon.HTTP_201
152			resp.location = '/emailservers/' + str(new_id)
153
154
155			class EmailServerItem:
156			def __init__(self):
157			pass
158
159			@staticmethod
160			def on_options(req, resp, id_):
161			_ = req
162			resp.status = falcon.HTTP_200
163			_ = id_
164
165			@staticmethod
166			def on_get(req, resp, id_):
167			admin_control(req)
168			if not id_.isdigit() or int(id_) <= 0:
169			raise falcon.HTTPError(status=falcon.HTTP_400, title='400 Bad Request')
170
171			cnx = mysql.connector.connect(**config.myems_fdd_db)
172			cursor = cnx.cursor()
173
174			query = (" SELECT id, host, port, requires_authentication, user_name, password, from_addr "
175			" FROM tbl_email_servers "
176			" WHERE id = %s ")
177			cursor.execute(query, (id_,))
178			row = cursor.fetchone()
179			cursor.close()
180			cnx.close()
181			if row is None:
182			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND')
183
184			result = {"id": row[0],
185			"host": row[1],
186			"port": row[2],
187			"requires_authentication": bool(row[3]),
188			"user_name": row[4],
189			"password": str(base64.b64decode(bytearray(row[5], 'utf-8')), 'utf-8')
190			if row[5] is not None else None,
191			"from_addr": row[5]}
192			resp.text = json.dumps(result)
193
194			@staticmethod
195			@user_logger
196			def on_delete(req, resp, id_):
197			"""Handles DELETE requests"""
198			admin_control(req)
199			if not id_.isdigit() or int(id_) <= 0:
200			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
201			description='API.INVALID_EMAIL_SERVER_ID')
202
203			cnx = mysql.connector.connect(**config.myems_fdd_db)
204			cursor = cnx.cursor()
205
206			cursor.execute(" SELECT host "
207			" FROM tbl_email_servers "
208			" WHERE id = %s ", (id_,))
209			if cursor.fetchone() is None:
210			cursor.close()
211			cnx.close()
212			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
213			description='API.EMAIL_SERVER_NOT_FOUND')
214
215			cursor.execute(" DELETE FROM tbl_email_servers WHERE id = %s ", (id_,))
216			cnx.commit()
217
218			cursor.close()
219			cnx.close()
220
221			resp.status = falcon.HTTP_204
222
223			@staticmethod
224			@user_logger
225			def on_put(req, resp, id_):
226			"""Handles PUT requests"""
227			admin_control(req)
228			try:
229			raw_json = req.stream.read().decode('utf-8')
230			except Exception as ex:
231			print(str(ex))
232			raise falcon.HTTPError(status=falcon.HTTP_400,
233			title='API.BAD_REQUEST',
234			description='API.FAILED_TO_READ_REQUEST_STREAM')
235
236			if not id_.isdigit() or int(id_) <= 0:
237			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
238			description='API.INVALID_EMAIL_SERVER_ID')
239
240			new_values = json.loads(raw_json)
241			if 'host' not in new_values['data'].keys() or \
242			not isinstance(new_values['data']['host'], str) or \
243			len(str.strip(new_values['data']['host'])) == 0:
244			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
245			description='API.INVALID_EMAIL_SERVER_HOST')
246
247			host = str.strip(new_values['data']['host'])
248
249			if 'port' not in new_values['data'].keys() or \
250			not isinstance(new_values['data']['port'], int) or \
251			new_values['data']['port'] <= 0:
252			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
253			description='API.INVALID_PORT')
254			port = float(new_values['data']['port'])
255
256			if 'requires_authentication' not in new_values['data'].keys() or \
257			not isinstance(new_values['data']['requires_authentication'], bool):
258			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
259			description='API.INVALID_REQUIRES_AUTHENTICATION')
260			requires_authentication = new_values['data']['requires_authentication']
261
262			if requires_authentication:
263			if 'user_name' not in new_values['data'].keys() or \
264			not isinstance(new_values['data']['user_name'], str) or \
265			len(str.strip(new_values['data']['user_name'])) == 0:
266			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
267			description='API.INVALID_USER_NAME')
268			user_name = new_values['data']['user_name']
269			else:
270			user_name = None
271
272			if requires_authentication:
273			if 'password' not in new_values['data'].keys() or \
274			not isinstance(new_values['data']['password'], str) or \
275			len(str.strip(new_values['data']['password'])) == 0:
276			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
277			description='API.INVALID_PASSWORD')
278			password = base64.b64encode(bytearray(new_values['data']['password'], 'utf-8'))
279			else:
280			password = None
281
282			if 'from_addr' not in new_values['data'].keys() or \
283			not isinstance(new_values['data']['from_addr'], str) or \
284			len(str.strip(new_values['data']['from_addr'])) == 0:
285			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
286			description='API.INVALID_FROM_ADDR')
287			from_addr = new_values['data']['from_addr']
288
289			match = re.match(r'^[_A-Za-z0-9-]+(\.[_A-Za-z0-9-]+)@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)(\.[A-Za-z]{2,4})$',
290			from_addr)
291			if match is None:
292			raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
293			description='API.INVALID_FROM_ADDR')
294
295			cnx = mysql.connector.connect(**config.myems_fdd_db)
296			cursor = cnx.cursor()
297
298			cursor.execute(" SELECT id "
299			" FROM tbl_email_servers "
300			" WHERE id = %s ",
301			(id_,))
302			if cursor.fetchone() is None:
303			cursor.close()
304			cnx.close()
305			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
306			description='API.EMAIL_SERVER_NOT_FOUND')
307
308			cursor.execute(" SELECT host "
309			" FROM tbl_email_servers "
310			" WHERE host = %s AND id != %s ", (host, id_))
311			if cursor.fetchone() is not None:
312			cursor.close()
313			cnx.close()
314			raise falcon.HTTPError(status=falcon.HTTP_404, title='API.BAD_REQUEST',
315			description='API.EMAIL_SERVER_HOST_IS_ALREADY_IN_USE')
316
317			update_row = (" UPDATE tbl_email_servers "
318			" SET host = %s, port = %s, requires_authentication = %s, "
319			" user_name = %s, password = %s, from_addr = %s "
320			" WHERE id = %s ")
321			cursor.execute(update_row, (host,
322			port,
323			requires_authentication,
324			user_name,
325			password,
326			from_addr,
327			id_,))
328			cnx.commit()
329
330			cursor.close()
331			cnx.close()
332
333			resp.status = falcon.HTTP_200
334

MyEMS / myems

core.emailserver.EmailServerItem.on_put() F last analyzed 2025-12-05 00:16 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

core.emailserver.EmailServerItem.on_put() F
last analyzed 2025-12-05 00:16 UTC