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

Passed

Push — master ( 08fcc0...b4637d )

by Guangyu

created 2021-11-10 10:33 UTC

core.useractivity.user_logger() F

↳ Parent: core.useractivity

Complexity

Conditions

Size

Total Lines	79
Code Lines	63

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	63
dl	0
loc	79
rs	1.8
c	0
b	0
f	0
cc	17
nop	1

How to fix Long Method Complexity

import os
from functools import wraps
import config
import mysql.connector
from datetime import datetime
import uuid
from gunicorn.http.body import Body
import simplejson as json
import falcon


def access_control(req):
    """
    Check administrator privilege in request headers to protect resources from invalid access
    :param req: HTTP request
    :return: HTTPError if invalid else None
    """
    if 'USER-UUID' not in req.headers or \
            not isinstance(req.headers['USER-UUID'], str) or \
            len(str.strip(req.headers['USER-UUID'])) == 0:
        raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                               description='API.INVALID_USER_UUID')
    admin_user_uuid = str.strip(req.headers['USER-UUID'])

    if 'TOKEN' not in req.headers or \
            not isinstance(req.headers['TOKEN'], str) or \
            len(str.strip(req.headers['TOKEN'])) == 0:
        raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                               description='API.INVALID_TOKEN')
    admin_token = str.strip(req.headers['TOKEN'])

    # Check administrator privilege
    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, (admin_user_uuid, admin_token,))
    row = cursor.fetchone()

    if row is None:

        cursor.close()
        cnx.disconnect()
        raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
                               description='API.ADMINISTRATOR_SESSION_NOT_FOUND')
    else:
        utc_expires = row[0]
        if datetime.utcnow() > utc_expires:
            cursor.close()
            cnx.disconnect()
            raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.ADMINISTRATOR_SESSION_TIMEOUT')
    query = (" SELECT name "
             " FROM tbl_users "
             " WHERE uuid = %s AND is_admin = true ")
    cursor.execute(query, (admin_user_uuid,))
    row = cursor.fetchone()
    cursor.close()
    cnx.disconnect()
    if row is None:
        raise falcon.HTTPError(falcon.HTTP_400, 'API.BAD_REQUEST', 'API.INVALID_PRIVILEGE')


def write_log(user_uuid, request_method, resource_type, resource_id, request_body):
    """
    :param user_uuid: user_uuid
    :param request_method: 'POST', 'PUT', 'DELETE'
    :param resource_type: class_name
    :param resource_id: int
    :param request_body: json in raw string
    """
    cnx = None
    cursor = None
    try:
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor()
        add_row = (" INSERT INTO tbl_logs "
                   "    (user_uuid, request_datetime_utc, request_method, resource_type, resource_id, request_body) "
                   " VALUES (%s, %s, %s, %s, %s , %s) ")
        cursor.execute(add_row, (user_uuid,
                                 datetime.utcnow(),
                                 request_method,
                                 resource_type,
                                 resource_id if resource_id else None,
                                 request_body if request_body else None,
                                 ))
        cnx.commit()
    except Exception as e:
        print('write_log:' + str(e))
    finally:
        if cnx:
            cnx.disconnect()
        if cursor:
            cursor.close()


def user_logger(func):
    """
    Decorator for logging user activities
    :param func: the decorated function
    :return: the decorator
    """
    @wraps(func)
    def logger(*args, **kwargs):
        qualified_name = func.__qualname__
        class_name = qualified_name.split(".")[0]
        func_name = qualified_name.split(".")[1]

        if func_name not in ("on_post", "on_put", "on_delete"):
            # do not log for other HTTP Methods
            func(*args, **kwargs)
            return
        req, resp = args
        headers = req.headers
        if headers is not None and 'USER-UUID' in headers.keys():
            user_uuid = headers['USER-UUID']
        else:
            # todo: deal with requests with NULL user_uuid
            print('user_logger: USER-UUID is NULL')
            # do not log for NULL user_uuid
            func(*args, **kwargs)
            return

        if func_name == "on_post":
            try:
                file_name = str(uuid.uuid4())
                with open(file_name, "wb") as fw:
                    reads = req.stream.read()
                    fw.write(reads)
                raw_json = reads.decode('utf-8')
                with open(file_name, "rb") as fr:
                    req.stream = Body(fr)
                    os.remove(file_name)
                    func(*args, **kwargs)
                    write_log(user_uuid=user_uuid, request_method='POST', resource_type=class_name,
                              resource_id=kwargs.get('id_'), request_body=raw_json)
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))
            return
        elif func_name == "on_put":
            try:
                file_name = str(uuid.uuid4())
                with open(file_name, "wb") as fw:
                    reads = req.stream.read()
                    fw.write(reads)
                raw_json = reads.decode('utf-8')
                with open(file_name, "rb") as fr:
                    req.stream = Body(fr)
                    os.remove(file_name)
                    func(*args, **kwargs)
                    write_log(user_uuid=user_uuid, request_method='PUT', resource_type=class_name,
                              resource_id=kwargs.get('id_'), request_body=raw_json)
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))

            return
        elif func_name == "on_delete":
            try:
                func(*args, **kwargs)
                write_log(user_uuid=user_uuid, request_method="DELETE", resource_type=class_name,
                          resource_id=kwargs.get('id_'), request_body=json.dumps(kwargs))
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))
            return

    return logger


1		import os
2		from functools import wraps
3		import config
4		import mysql.connector
5		from datetime import datetime
6		import uuid
7		from gunicorn.http.body import Body
8		import simplejson as json
9		import falcon
10
11
12		def access_control(req):
13		"""
14		Check administrator privilege in request headers to protect resources from invalid access
15		:param req: HTTP request
16		:return: HTTPError if invalid else None
17		"""
18		if 'USER-UUID' not in req.headers or \
19		not isinstance(req.headers['USER-UUID'], str) or \
20		len(str.strip(req.headers['USER-UUID'])) == 0:
21		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
22		description='API.INVALID_USER_UUID')
23		admin_user_uuid = str.strip(req.headers['USER-UUID'])
24
25		if 'TOKEN' not in req.headers or \
26		not isinstance(req.headers['TOKEN'], str) or \
27		len(str.strip(req.headers['TOKEN'])) == 0:
28		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
29		description='API.INVALID_TOKEN')
30		admin_token = str.strip(req.headers['TOKEN'])
31
32		# Check administrator privilege
33		cnx = mysql.connector.connect(**config.myems_user_db)
34		cursor = cnx.cursor()
35		query = (" SELECT utc_expires "
36		" FROM tbl_sessions "
37		" WHERE user_uuid = %s AND token = %s")
38		cursor.execute(query, (admin_user_uuid, admin_token,))
39		row = cursor.fetchone()
40
41	View Code Duplication	if row is None:
		0 ignored issues – show Duplication introduced 2021-11-10 08:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
42		cursor.close()
43		cnx.disconnect()
44		raise falcon.HTTPError(falcon.HTTP_404, title='API.NOT_FOUND',
45		description='API.ADMINISTRATOR_SESSION_NOT_FOUND')
46		else:
47		utc_expires = row[0]
48		if datetime.utcnow() > utc_expires:
49		cursor.close()
50		cnx.disconnect()
51		raise falcon.HTTPError(falcon.HTTP_400, title='API.BAD_REQUEST',
52		description='API.ADMINISTRATOR_SESSION_TIMEOUT')
53		query = (" SELECT name "
54		" FROM tbl_users "
55		" WHERE uuid = %s AND is_admin = true ")
56		cursor.execute(query, (admin_user_uuid,))
57		row = cursor.fetchone()
58		cursor.close()
59		cnx.disconnect()
60		if row is None:
61		raise falcon.HTTPError(falcon.HTTP_400, 'API.BAD_REQUEST', 'API.INVALID_PRIVILEGE')
62
63
64		def write_log(user_uuid, request_method, resource_type, resource_id, request_body):
65		"""
66		:param user_uuid: user_uuid
67		:param request_method: 'POST', 'PUT', 'DELETE'
68		:param resource_type: class_name
69		:param resource_id: int
70		:param request_body: json in raw string
71		"""
72		cnx = None
73		cursor = None
74		try:
75		cnx = mysql.connector.connect(**config.myems_user_db)
76		cursor = cnx.cursor()
77		add_row = (" INSERT INTO tbl_logs "
78		" (user_uuid, request_datetime_utc, request_method, resource_type, resource_id, request_body) "
79		" VALUES (%s, %s, %s, %s, %s , %s) ")
80		cursor.execute(add_row, (user_uuid,
81		datetime.utcnow(),
82		request_method,
83		resource_type,
84		resource_id if resource_id else None,
85		request_body if request_body else None,
86		))
87		cnx.commit()
88		except Exception as e:
89		print('write_log:' + str(e))
90		finally:
91		if cnx:
92		cnx.disconnect()
93		if cursor:
94		cursor.close()
95
96
97		def user_logger(func):
98		"""
99		Decorator for logging user activities
100		:param func: the decorated function
101		:return: the decorator
102		"""
103		@wraps(func)
104		def logger(args, *kwargs):
105		qualified_name = func.__qualname__
106		class_name = qualified_name.split(".")[0]
107		func_name = qualified_name.split(".")[1]
108
109		if func_name not in ("on_post", "on_put", "on_delete"):
110		# do not log for other HTTP Methods
111		func(args, *kwargs)
112		return
113		req, resp = args
114		headers = req.headers
115		if headers is not None and 'USER-UUID' in headers.keys():
116		user_uuid = headers['USER-UUID']
117		else:
118		# todo: deal with requests with NULL user_uuid
119		print('user_logger: USER-UUID is NULL')
120		# do not log for NULL user_uuid
121		func(args, *kwargs)
122		return
123
124		if func_name == "on_post":
125		try:
126		file_name = str(uuid.uuid4())
127		with open(file_name, "wb") as fw:
128		reads = req.stream.read()
129		fw.write(reads)
130		raw_json = reads.decode('utf-8')
131		with open(file_name, "rb") as fr:
132		req.stream = Body(fr)
133		os.remove(file_name)
134		func(args, *kwargs)
135		write_log(user_uuid=user_uuid, request_method='POST', resource_type=class_name,
136		resource_id=kwargs.get('id_'), request_body=raw_json)
137		except Exception as e:
138		if isinstance(e, falcon.HTTPError):
139		raise e
140		else:
141		print('user_logger:' + str(e))
142		return
143		elif func_name == "on_put":
144		try:
145		file_name = str(uuid.uuid4())
146		with open(file_name, "wb") as fw:
147		reads = req.stream.read()
148		fw.write(reads)
149		raw_json = reads.decode('utf-8')
150		with open(file_name, "rb") as fr:
151		req.stream = Body(fr)
152		os.remove(file_name)
153		func(args, *kwargs)
154		write_log(user_uuid=user_uuid, request_method='PUT', resource_type=class_name,
155		resource_id=kwargs.get('id_'), request_body=raw_json)
156		except Exception as e:
157		if isinstance(e, falcon.HTTPError):
158		raise e
159		else:
160		print('user_logger:' + str(e))
161
162		return
163		elif func_name == "on_delete":
164		try:
165		func(args, *kwargs)
166		write_log(user_uuid=user_uuid, request_method="DELETE", resource_type=class_name,
167		resource_id=kwargs.get('id_'), request_body=json.dumps(kwargs))
168		except Exception as e:
169		if isinstance(e, falcon.HTTPError):
170		raise e
171		else:
172		print('user_logger:' + str(e))
173		return
174
175		return logger
176

MyEMS / myems

Push — master ( 08fcc0...b4637d )

core.useractivity.user_logger() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like