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

Passed

Push — master ( 47b091...6083b9 )

by Guangyu

created 2021-02-14 10:07 UTC

NotificationCollection.init() A

↳ Parent: core.notification

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import falcon
import json
import mysql.connector
import config
from datetime import datetime, timedelta, timezone


class NotificationCollection:
    @staticmethod
    def __init__():
        pass

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

    @staticmethod
    def on_get(req, resp):
        status = req.params.get('status')
        start_datetime_local = req.params.get('startdatetime')
        end_datetime_local = req.params.get('enddatetime')
        if status is not None:
            status = str.strip(status)
            if status not in ['new', 'read', 'archived']:
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_STATUS')

        timezone_offset = int(config.utc_offset[1:3]) * 60 + int(config.utc_offset[4:6])
        if config.utc_offset[0] == '-':
            timezone_offset = -timezone_offset
        start_datetime_utc = None
        if start_datetime_local is not None and len(str.strip(start_datetime_local)) > 0:
            start_datetime_local = str.strip(start_datetime_local)
            try:
                start_datetime_utc = datetime.strptime(start_datetime_local,
                                                       '%Y-%m-%dT%H:%M:%S').replace(tzinfo=timezone.utc) - \
                                     timedelta(minutes=timezone_offset)
            except ValueError:
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description="API.INVALID_START_DATETIME")

        end_datetime_utc = None
        if end_datetime_local is not None and len(str.strip(end_datetime_local)) > 0:
            end_datetime_local = str.strip(end_datetime_local)
            try:
                end_datetime_utc = datetime.strptime(end_datetime_local,
                                                     '%Y-%m-%dT%H:%M:%S').replace(tzinfo=timezone.utc) - \
                                        timedelta(minutes=timezone_offset)
            except ValueError:
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description="API.INVALID_END_DATETIME")

        if start_datetime_utc is not None and end_datetime_utc is not None and \
                start_datetime_utc >= end_datetime_utc:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_END_DATETIME')

        # Verify User Session
        cookie_dict = dict()
        cookies = req.headers['COOKIE'].split(';')
        for cookie in cookies:
            kv = cookie.split('=')
            cookie_dict[str.strip(kv[0])] = str.strip(kv[1])

        if cookie_dict.get('user_uuid') is None or cookie_dict.get('token') is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COOKIES_PLEASE_RE_LOGIN')

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

        query = (" SELECT utc_expires "
                 " FROM tbl_sessions "
                 " WHERE user_uuid = %s AND token = %s")
        cursor.execute(query, (cookie_dict.get('user_uuid'), cookie_dict.get('token'),))
        row = cursor.fetchone()

        if row is None:

            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
        else:
            utc_expires = row[0]
            if datetime.utcnow() > utc_expires:
                if cursor:
                    cursor.close()
                if cnx:
                    cnx.disconnect()
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE uuid = %s ",
                       (cookie_dict.get('user_uuid'),))
        row = cursor.fetchone()
        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')
        else:
            user_id = row[0]

        # get notifications
        if status is None:
            query = (" SELECT id, created_datetime_utc, status, subject, message, url "
                     " FROM tbl_notifications "
                     " WHERE user_id = %s AND "
                     "       created_datetime_utc >= %s AND created_datetime_utc < %s "
                     " ORDER BY created_datetime_utc DESC ")
            cursor.execute(query, (user_id, start_datetime_utc, end_datetime_utc))
        else:
            query = (" SELECT id, created_datetime_utc, status, subject, message, url "
                     " FROM tbl_notifications "
                     " WHERE user_id = %s AND "
                     "       created_datetime_utc >= %s AND created_datetime_utc < %s AND "
                     "       status = %s "
                     " ORDER BY created_datetime_utc DESC ")
            cursor.execute(query, (user_id, start_datetime_utc, end_datetime_utc, status))
        rows = cursor.fetchall()

        if cursor:
            cursor.close()
        if cnx:
            cnx.disconnect()

        result = list()
        if rows is not None and len(rows) > 0:
            for row in rows:
                meta_result = {"id": row[0],
                               "created_datetime":
                                   (row[1] +
                                    timedelta(hours=int(config.utc_offset[1:3]))).strftime('%Y-%m-%dT%H:%M:%S'),
                               "status": row[2],
                               "subject": row[3],
                               "message": row[4],
                               "url": row[5]}
                result.append(meta_result)

        resp.body = json.dumps(result)


class NotificationItem:
    @staticmethod
    def __init__():
        pass

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

    @staticmethod
    def on_get(req, resp, id_):
        """Handles GET requests"""
        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_NOTIFICATION_ID')

        # Verify User Session
        cookie_dict = dict()
        cookies = req.headers['COOKIE'].split(';')
        for cookie in cookies:
            kv = cookie.split('=')
            cookie_dict[str.strip(kv[0])] = str.strip(kv[1])

        if cookie_dict.get('user_uuid') is None or cookie_dict.get('token') is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COOKIES_PLEASE_RE_LOGIN')

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

        query = (" SELECT utc_expires "
                 " FROM tbl_sessions "
                 " WHERE user_uuid = %s AND token = %s")
        cursor.execute(query, (cookie_dict.get('user_uuid'), cookie_dict.get('token'),))
        row = cursor.fetchone()

        if row is None:

            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
        else:
            utc_expires = row[0]
            if datetime.utcnow() > utc_expires:
                if cursor:
                    cursor.close()
                if cnx:
                    cnx.disconnect()
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE uuid = %s ",
                       (cookie_dict.get('user_uuid'),))
        row = cursor.fetchone()
        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')
        else:
            user_id = row[0]

        # get notification
        query = (" SELECT id, created_datetime_utc, status, subject, message, url "
                 " FROM tbl_notifications "
                 " WHERE id = %s AND user_id = %s ")
        cursor.execute(query, (id_, user_id))
        row = cursor.fetchone()

        if cursor:
            cursor.close()
        if cnx:
            cnx.disconnect()

        if row is None:
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.NOTIFICATION_NOT_FOUND')

        meta_result = {"id": row[0],
                       "created_datetime":
                           (row[1] +
                            timedelta(hours=int(config.utc_offset[1:3]))).strftime('%Y-%m-%dT%H:%M:%S'),
                       "status": row[2],
                       "subject": row[3],
                       "message": row[4],
                       "url": row[5]}

        resp.body = json.dumps(meta_result)

    @staticmethod
    def on_put(req, resp, id_):
        """Handles PUT requests"""
        try:
            raw_json = req.stream.read().decode('utf-8')
        except Exception as ex:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.EXCEPTION', description=ex)

        if not id_.isdigit() or int(id_) <= 0:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_NOTIFICATION_ID')

        new_values = json.loads(raw_json)

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

        # Verify User Session
        cookie_dict = dict()
        cookies = req.headers['COOKIE'].split(';')
        for cookie in cookies:
            kv = cookie.split('=')
            cookie_dict[str.strip(kv[0])] = str.strip(kv[1])

        if cookie_dict.get('user_uuid') is None or cookie_dict.get('token') is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COOKIES_PLEASE_RE_LOGIN')

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

        query = (" SELECT utc_expires "
                 " FROM tbl_sessions "
                 " WHERE user_uuid = %s AND token = %s")
        cursor.execute(query, (cookie_dict.get('user_uuid'), cookie_dict.get('token'),))
        row = cursor.fetchone()

        if row is None:

            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
        else:
            utc_expires = row[0]
            if datetime.utcnow() > utc_expires:
                if cursor:
                    cursor.close()
                if cnx:
                    cnx.disconnect()
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE uuid = %s ",
                       (cookie_dict.get('user_uuid'),))
        row = cursor.fetchone()
        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')
        else:
            user_id = row[0]

        cursor.execute(" SELECT id "
                       " FROM tbl_notifications "
                       " WHERE id = %s AND user_id = %s ", (id_, user_id))
        if cursor.fetchone() is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.NOTIFICATION_NOT_FOUND')

        update_row = (" UPDATE tbl_notifications "
                      " SET status = %s "
                      " WHERE id = %s ")
        cursor.execute(update_row, (status,
                                    id_,))
        cnx.commit()
        if cursor:
            cursor.close()
        if cnx:
            cnx.disconnect()

        resp.status = falcon.HTTP_200

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

        # Verify User Session
        cookie_dict = dict()
        cookies = req.headers['COOKIE'].split(';')
        for cookie in cookies:
            kv = cookie.split('=')
            cookie_dict[str.strip(kv[0])] = str.strip(kv[1])

        if cookie_dict.get('user_uuid') is None or cookie_dict.get('token') is None:
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_COOKIES_PLEASE_RE_LOGIN')

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

        query = (" SELECT utc_expires "
                 " FROM tbl_sessions "
                 " WHERE user_uuid = %s AND token = %s")
        cursor.execute(query, (cookie_dict.get('user_uuid'), cookie_dict.get('token'),))
        row = cursor.fetchone()

        if row is None:

            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_SESSION_PLEASE_RE_LOGIN')
        else:
            utc_expires = row[0]
            if datetime.utcnow() > utc_expires:
                if cursor:
                    cursor.close()
                if cnx:
                    cnx.disconnect()
                raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                       description='API.USER_SESSION_TIMEOUT')

        cursor.execute(" SELECT id "
                       " FROM tbl_users "
                       " WHERE uuid = %s ",
                       (cookie_dict.get('user_uuid'),))
        row = cursor.fetchone()
        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.INVALID_USER_PLEASE_RE_LOGIN')
        else:
            user_id = row[0]

        cursor.execute(" SELECT id "
                       " FROM tbl_notifications "
                       " WHERE id = %s AND user_id = %s ", (id_, user_id))
        row = cursor.fetchone()

        if row is None:
            if cursor:
                cursor.close()
            if cnx:
                cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                                   description='API.NOTIFICATION_NOT_FOUND')

        cursor.execute(" DELETE FROM tbl_notifications WHERE id = %s ", (id_,))
        cnx.commit()
        if cursor:
            cursor.close()
        if cnx:
            cnx.disconnect()

        resp.status = falcon.HTTP_204


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

myems / myems-api

Push — master ( 47b091...6083b9 )

NotificationCollection.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

NotificationCollection.init() A