Code Duplication - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 89-94 lines in 2 locations

myems-api/core/sensor.py 1 location


import config


class SensorCollection:
    def __init__(self):
        """"Initializes SensorCollection"""
        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)
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid, description "
                 " FROM tbl_sensors "
                 " ORDER BY id ")
        cursor.execute(query)
        rows_sensors = cursor.fetchall()

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

        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_sensors "
                       " 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.SENSOR_NAME_IS_ALREADY_IN_USE')

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

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


class SensorItem:

myems-api/core/workingcalendar.py 1 location


import config


class WorkingCalendarCollection:
    def __init__(self):
        """"Initializes WorkingCalendarCollection"""
        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)
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        cursor.execute(" SELECT id, name, description"
                       " FROM tbl_working_calendars ")

        rows_calendars = cursor.fetchall()

        result = list()
        if rows_calendars is not None and len(rows_calendars) > 0:
            for row in rows_calendars:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "description": row[2]}
                result.append(meta_result)
    
        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

    @staticmethod
    def on_post(req, resp):
        """Handles POST requests"""
        admin_control(req)
        try:
            raw_json = req.stream.read().decode('utf-8')
            new_values = json.loads(raw_json)
        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 '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_WORKING_CALENDAR_NAME')
        name = str.strip(new_values['data']['name'])

        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_working_calendars "
                       " 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.WORKING_CALENDAR_NAME_IS_ALREADY_IN_USE')

        add_values = (" INSERT INTO tbl_working_calendars "
                      " (name, description) "
                      " VALUES (%s, %s) ")
        cursor.execute(add_values, (name,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class WorkingCalendarItem:

		@@ 10-103 (lines=94) @@
7		import config
8
9
10		class SensorCollection:
11		def __init__(self):
12		""""Initializes SensorCollection"""
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		cnx = mysql.connector.connect(**config.myems_system_db)
29		cursor = cnx.cursor()
30
31		query = (" SELECT id, name, uuid, description "
32		" FROM tbl_sensors "
33		" ORDER BY id ")
34		cursor.execute(query)
35		rows_sensors = cursor.fetchall()
36
37		result = list()
38		if rows_sensors is not None and len(rows_sensors) > 0:
39		for row in rows_sensors:
40		meta_result = {"id": row[0],
41		"name": row[1],
42		"uuid": row[2],
43		"description": row[3]}
44		result.append(meta_result)
45
46		cursor.close()
47		cnx.close()
48		resp.text = json.dumps(result)
49
50		@staticmethod
51		@user_logger
52		def on_post(req, resp):
53		"""Handles POST requests"""
54		admin_control(req)
55		try:
56		raw_json = req.stream.read().decode('utf-8')
57		except Exception as ex:
58		print(str(ex))
59		raise falcon.HTTPError(status=falcon.HTTP_400,
60		title='API.BAD_REQUEST',
61		description='API.FAILED_TO_READ_REQUEST_STREAM')
62
63		new_values = json.loads(raw_json)
64
65		if 'name' not in new_values['data'].keys() or \
66		not isinstance(new_values['data']['name'], str) or \
67		len(str.strip(new_values['data']['name'])) == 0:
68		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
69		description='API.INVALID_SENSOR_NAME')
70		name = str.strip(new_values['data']['name'])
71
72		if 'description' in new_values['data'].keys() and \
73		new_values['data']['description'] is not None and \
74		len(str(new_values['data']['description'])) > 0:
75		description = str.strip(new_values['data']['description'])
76		else:
77		description = None
78
79		cnx = mysql.connector.connect(**config.myems_system_db)
80		cursor = cnx.cursor()
81
82		cursor.execute(" SELECT name "
83		" FROM tbl_sensors "
84		" WHERE name = %s ", (name,))
85		if cursor.fetchone() is not None:
86		cursor.close()
87		cnx.close()
88		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
89		description='API.SENSOR_NAME_IS_ALREADY_IN_USE')
90
91		add_values = (" INSERT INTO tbl_sensors "
92		" (name, uuid, description) "
93		" VALUES (%s, %s, %s) ")
94		cursor.execute(add_values, (name,
95		str(uuid.uuid4()),
96		description))
97		new_id = cursor.lastrowid
98		cnx.commit()
99		cursor.close()
100		cnx.close()
101
102		resp.status = falcon.HTTP_201
103		resp.location = '/sensors/' + str(new_id)
104
105
106		class SensorItem:

		@@ 9-97 (lines=89) @@
6		import config
7
8
9		class WorkingCalendarCollection:
10		def __init__(self):
11		""""Initializes WorkingCalendarCollection"""
12		pass
13
14		@staticmethod
15		def on_options(req, resp):
16		_ = req
17		resp.status = falcon.HTTP_200
18
19		@staticmethod
20		def on_get(req, resp):
21		if 'API-KEY' not in req.headers or \
22		not isinstance(req.headers['API-KEY'], str) or \
23		len(str.strip(req.headers['API-KEY'])) == 0:
24		access_control(req)
25		else:
26		api_key_control(req)
27		cnx = mysql.connector.connect(**config.myems_system_db)
28		cursor = cnx.cursor()
29
30		cursor.execute(" SELECT id, name, description"
31		" FROM tbl_working_calendars ")
32
33		rows_calendars = cursor.fetchall()
34
35		result = list()
36		if rows_calendars is not None and len(rows_calendars) > 0:
37		for row in rows_calendars:
38		meta_result = {"id": row[0],
39		"name": row[1],
40		"description": row[2]}
41		result.append(meta_result)
42
43		cursor.close()
44		cnx.close()
45		resp.text = json.dumps(result)
46
47		@staticmethod
48		def on_post(req, resp):
49		"""Handles POST requests"""
50		admin_control(req)
51		try:
52		raw_json = req.stream.read().decode('utf-8')
53		new_values = json.loads(raw_json)
54		except Exception as ex:
55		print(str(ex))
56		raise falcon.HTTPError(status=falcon.HTTP_400,
57		title='API.BAD_REQUEST',
58		description='API.FAILED_TO_READ_REQUEST_STREAM')
59
60		if 'name' not in new_values['data'].keys() or \
61		not isinstance(new_values['data']['name'], str) or \
62		len(str.strip(new_values['data']['name'])) == 0:
63		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
64		description='API.INVALID_WORKING_CALENDAR_NAME')
65		name = str.strip(new_values['data']['name'])
66
67		if 'description' in new_values['data'].keys() and \
68		new_values['data']['description'] is not None and \
69		len(str(new_values['data']['description'])) > 0:
70		description = str.strip(new_values['data']['description'])
71		else:
72		description = None
73
74		cnx = mysql.connector.connect(**config.myems_system_db)
75		cursor = cnx.cursor()
76
77		cursor.execute(" SELECT name "
78		" FROM tbl_working_calendars "
79		" WHERE name = %s ", (name,))
80		if cursor.fetchone() is not None:
81		cursor.close()
82		cnx.close()
83		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
84		description='API.WORKING_CALENDAR_NAME_IS_ALREADY_IN_USE')
85
86		add_values = (" INSERT INTO tbl_working_calendars "
87		" (name, description) "
88		" VALUES (%s, %s) ")
89		cursor.execute(add_values, (name,
90		description))
91		new_id = cursor.lastrowid
92		cnx.commit()
93		cursor.close()
94		cnx.close()
95
96		resp.status = falcon.HTTP_201
97		resp.location = '/workingcalendar/' + str(new_id)
98
99
100		class WorkingCalendarItem:

MyEMS / myems

Code Duplication Length = 89-94 lines in 2 locations

myems-api/core/sensor.py 1 location

myems-api/core/workingcalendar.py 1 location