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

Code Duplication Length = 188-189 lines in 2 locations

myems-api/core/equipment.py 1 location


import config


class EquipmentCollection:
    def __init__(self):
        """Initializes EquipmentCollection"""
        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 "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()

        svg_dict = dict()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        is_input_counted, is_output_counted, "
                 "        cost_center_id, svg_id, camera_url, description "
                 " FROM tbl_equipments "
                 " ORDER BY id ")
        cursor.execute(query)
        rows_equipments = cursor.fetchall()

        result = list()
        if rows_equipments is not None and len(rows_equipments) > 0:
            for row in rows_equipments:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "is_input_counted": bool(row[3]),
                               "is_output_counted": bool(row[4]),
                               "cost_center": cost_center_dict.get(row[5], None),
                               "svg": svg_dict.get(row[6], None),
                               "camera_url": row[7],
                               "description": row[8],
                               "qrcode": 'equipment:' + row[2]}
                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_EQUIPMENT_NAME')
        name = str.strip(new_values['data']['name'])

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

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

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

        if 'svg_id' in new_values['data'].keys() and \
                isinstance(new_values['data']['svg_id'], int) and \
                new_values['data']['svg_id'] > 0:
            svg_id = new_values['data']['svg_id']
        else:
            svg_id = None

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

        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_equipments "
                       " 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.EQUIPMENT_NAME_IS_ALREADY_IN_USE')

        if cost_center_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_cost_centers "
                           " WHERE id = %s ",
                           (new_values['data']['cost_center_id'],))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.COST_CENTER_NOT_FOUND')

        if svg_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_svgs "
                           " WHERE id = %s ",
                           (svg_id,))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.SVG_NOT_FOUND')

        add_values = (" INSERT INTO tbl_equipments "
                      "    (name, uuid, is_input_counted, is_output_counted, "
                      "     cost_center_id, svg_id, camera_url, description) "
                      " VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    is_input_counted,
                                    is_output_counted,
                                    cost_center_id,
                                    svg_id,
                                    camera_url,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class EquipmentItem:

myems-api/core/combinedequipment.py 1 location


import config


class CombinedEquipmentCollection:
    def __init__(self):
        """ Initializes CombinedEquipmentCollection"""
        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 "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}
        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()

        svg_dict = dict()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        is_input_counted, is_output_counted, "
                 "        cost_center_id, svg_id, camera_url, description "
                 " FROM tbl_combined_equipments "
                 " ORDER BY id ")
        cursor.execute(query)
        rows_combined_equipments = cursor.fetchall()

        result = list()
        if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
            for row in rows_combined_equipments:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "is_input_counted": bool(row[3]),
                               "is_output_counted": bool(row[4]),
                               "cost_center": cost_center_dict.get(row[5], None),
                               "svg": svg_dict.get(row[6], None),
                               "camera_url": row[7],
                               "description": row[8],
                               "qrcode": 'combinedequipment:' + row[2]}
                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(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_COMBINED_EQUIPMENT_NAME')
        name = str.strip(new_values['data']['name'])

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

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

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

        if 'svg_id' in new_values['data'].keys() and \
                isinstance(new_values['data']['svg_id'], int) and \
                new_values['data']['svg_id'] > 0:
            svg_id = new_values['data']['svg_id']
        else:
            svg_id = None

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

        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_combined_equipments "
                       " 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.COMBINED_EQUIPMENT_NAME_IS_ALREADY_IN_USE')

        if cost_center_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_cost_centers "
                           " WHERE id = %s ",
                           (new_values['data']['cost_center_id'],))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.COST_CENTER_NOT_FOUND')

        if svg_id is not None:
            cursor.execute(" SELECT name "
                           " FROM tbl_svgs "
                           " WHERE id = %s ",
                           (svg_id,))
            row = cursor.fetchone()
            if row is None:
                cursor.close()
                cnx.close()
                raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                                       description='API.SVG_NOT_FOUND')

        add_values = (" INSERT INTO tbl_combined_equipments "
                      "    (name, uuid, is_input_counted, is_output_counted, "
                      "     cost_center_id, svg_id, camera_url, description) "
                      " VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ")
        cursor.execute(add_values, (name,
                                    str(uuid.uuid4()),
                                    is_input_counted,
                                    is_output_counted,
                                    cost_center_id,
                                    svg_id,
                                    camera_url,
                                    description))
        new_id = cursor.lastrowid
        cnx.commit()
        cursor.close()
        cnx.close()

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


class CombinedEquipmentItem:

		@@ 10-198 (lines=189) @@
7		import config
8
9
10		class EquipmentCollection:
11		def __init__(self):
12		"""Initializes EquipmentCollection"""
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 "
32		" FROM tbl_cost_centers ")
33		cursor.execute(query)
34		rows_cost_centers = cursor.fetchall()
35
36		cost_center_dict = dict()
37		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
38		for row in rows_cost_centers:
39		cost_center_dict[row[0]] = {"id": row[0],
40		"name": row[1],
41		"uuid": row[2]}
42
43		query = (" SELECT id, name, uuid "
44		" FROM tbl_svgs ")
45		cursor.execute(query)
46		rows_svgs = cursor.fetchall()
47
48		svg_dict = dict()
49		if rows_svgs is not None and len(rows_svgs) > 0:
50		for row in rows_svgs:
51		svg_dict[row[0]] = {"id": row[0],
52		"name": row[1],
53		"uuid": row[2]}
54
55		query = (" SELECT id, name, uuid, "
56		" is_input_counted, is_output_counted, "
57		" cost_center_id, svg_id, camera_url, description "
58		" FROM tbl_equipments "
59		" ORDER BY id ")
60		cursor.execute(query)
61		rows_equipments = cursor.fetchall()
62
63		result = list()
64		if rows_equipments is not None and len(rows_equipments) > 0:
65		for row in rows_equipments:
66		meta_result = {"id": row[0],
67		"name": row[1],
68		"uuid": row[2],
69		"is_input_counted": bool(row[3]),
70		"is_output_counted": bool(row[4]),
71		"cost_center": cost_center_dict.get(row[5], None),
72		"svg": svg_dict.get(row[6], None),
73		"camera_url": row[7],
74		"description": row[8],
75		"qrcode": 'equipment:' + row[2]}
76		result.append(meta_result)
77
78		cursor.close()
79		cnx.close()
80		resp.text = json.dumps(result)
81
82		@staticmethod
83		@user_logger
84		def on_post(req, resp):
85		"""Handles POST requests"""
86		admin_control(req)
87		try:
88		raw_json = req.stream.read().decode('utf-8')
89		except Exception as ex:
90		print(str(ex))
91		raise falcon.HTTPError(status=falcon.HTTP_400,
92		title='API.BAD_REQUEST',
93		description='API.FAILED_TO_READ_REQUEST_STREAM')
94
95		new_values = json.loads(raw_json)
96
97		if 'name' not in new_values['data'].keys() or \
98		not isinstance(new_values['data']['name'], str) or \
99		len(str.strip(new_values['data']['name'])) == 0:
100		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
101		description='API.INVALID_EQUIPMENT_NAME')
102		name = str.strip(new_values['data']['name'])
103
104		if 'is_input_counted' not in new_values['data'].keys() or \
105		not isinstance(new_values['data']['is_input_counted'], bool):
106		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
107		description='API.INVALID_IS_INPUT_COUNTED_VALUE')
108		is_input_counted = new_values['data']['is_input_counted']
109
110		if 'is_output_counted' not in new_values['data'].keys() or \
111		not isinstance(new_values['data']['is_output_counted'], bool):
112		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
113		description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
114		is_output_counted = new_values['data']['is_output_counted']
115
116		if 'cost_center_id' not in new_values['data'].keys() or \
117		not isinstance(new_values['data']['cost_center_id'], int) or \
118		new_values['data']['cost_center_id'] <= 0:
119		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
120		description='API.INVALID_COST_CENTER_ID')
121		cost_center_id = new_values['data']['cost_center_id']
122
123		if 'svg_id' in new_values['data'].keys() and \
124		isinstance(new_values['data']['svg_id'], int) and \
125		new_values['data']['svg_id'] > 0:
126		svg_id = new_values['data']['svg_id']
127		else:
128		svg_id = None
129
130		if 'camera_url' in new_values['data'].keys() and \
131		new_values['data']['camera_url'] is not None and \
132		len(str(new_values['data']['camera_url'])) > 0:
133		camera_url = str.strip(new_values['data']['camera_url'])
134		else:
135		camera_url = None
136
137		if 'description' in new_values['data'].keys() and \
138		new_values['data']['description'] is not None and \
139		len(str(new_values['data']['description'])) > 0:
140		description = str.strip(new_values['data']['description'])
141		else:
142		description = None
143
144		cnx = mysql.connector.connect(**config.myems_system_db)
145		cursor = cnx.cursor()
146
147		cursor.execute(" SELECT name "
148		" FROM tbl_equipments "
149		" WHERE name = %s ", (name,))
150		if cursor.fetchone() is not None:
151		cursor.close()
152		cnx.close()
153		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
154		description='API.EQUIPMENT_NAME_IS_ALREADY_IN_USE')
155
156		if cost_center_id is not None:
157		cursor.execute(" SELECT name "
158		" FROM tbl_cost_centers "
159		" WHERE id = %s ",
160		(new_values['data']['cost_center_id'],))
161		row = cursor.fetchone()
162		if row is None:
163		cursor.close()
164		cnx.close()
165		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
166		description='API.COST_CENTER_NOT_FOUND')
167
168		if svg_id is not None:
169		cursor.execute(" SELECT name "
170		" FROM tbl_svgs "
171		" WHERE id = %s ",
172		(svg_id,))
173		row = cursor.fetchone()
174		if row is None:
175		cursor.close()
176		cnx.close()
177		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
178		description='API.SVG_NOT_FOUND')
179
180		add_values = (" INSERT INTO tbl_equipments "
181		" (name, uuid, is_input_counted, is_output_counted, "
182		" cost_center_id, svg_id, camera_url, description) "
183		" VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ")
184		cursor.execute(add_values, (name,
185		str(uuid.uuid4()),
186		is_input_counted,
187		is_output_counted,
188		cost_center_id,
189		svg_id,
190		camera_url,
191		description))
192		new_id = cursor.lastrowid
193		cnx.commit()
194		cursor.close()
195		cnx.close()
196
197		resp.status = falcon.HTTP_201
198		resp.location = '/equipments/' + str(new_id)
199
200
201		class EquipmentItem:

		@@ 10-197 (lines=188) @@
7		import config
8
9
10		class CombinedEquipmentCollection:
11		def __init__(self):
12		""" Initializes CombinedEquipmentCollection"""
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 "
32		" FROM tbl_cost_centers ")
33		cursor.execute(query)
34		rows_cost_centers = cursor.fetchall()
35
36		cost_center_dict = dict()
37		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
38		for row in rows_cost_centers:
39		cost_center_dict[row[0]] = {"id": row[0],
40		"name": row[1],
41		"uuid": row[2]}
42		query = (" SELECT id, name, uuid "
43		" FROM tbl_svgs ")
44		cursor.execute(query)
45		rows_svgs = cursor.fetchall()
46
47		svg_dict = dict()
48		if rows_svgs is not None and len(rows_svgs) > 0:
49		for row in rows_svgs:
50		svg_dict[row[0]] = {"id": row[0],
51		"name": row[1],
52		"uuid": row[2]}
53
54		query = (" SELECT id, name, uuid, "
55		" is_input_counted, is_output_counted, "
56		" cost_center_id, svg_id, camera_url, description "
57		" FROM tbl_combined_equipments "
58		" ORDER BY id ")
59		cursor.execute(query)
60		rows_combined_equipments = cursor.fetchall()
61
62		result = list()
63		if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
64		for row in rows_combined_equipments:
65		meta_result = {"id": row[0],
66		"name": row[1],
67		"uuid": row[2],
68		"is_input_counted": bool(row[3]),
69		"is_output_counted": bool(row[4]),
70		"cost_center": cost_center_dict.get(row[5], None),
71		"svg": svg_dict.get(row[6], None),
72		"camera_url": row[7],
73		"description": row[8],
74		"qrcode": 'combinedequipment:' + row[2]}
75		result.append(meta_result)
76
77		cursor.close()
78		cnx.close()
79		resp.text = json.dumps(result)
80
81		@staticmethod
82		@user_logger
83		def on_post(req, resp):
84		"""Handles POST requests"""
85		admin_control(req)
86		try:
87		raw_json = req.stream.read().decode('utf-8')
88		except Exception as ex:
89		print(ex)
90		raise falcon.HTTPError(status=falcon.HTTP_400,
91		title='API.BAD_REQUEST',
92		description='API.FAILED_TO_READ_REQUEST_STREAM')
93
94		new_values = json.loads(raw_json)
95
96		if 'name' not in new_values['data'].keys() or \
97		not isinstance(new_values['data']['name'], str) or \
98		len(str.strip(new_values['data']['name'])) == 0:
99		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
100		description='API.INVALID_COMBINED_EQUIPMENT_NAME')
101		name = str.strip(new_values['data']['name'])
102
103		if 'is_input_counted' not in new_values['data'].keys() or \
104		not isinstance(new_values['data']['is_input_counted'], bool):
105		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
106		description='API.INVALID_IS_INPUT_COUNTED_VALUE')
107		is_input_counted = new_values['data']['is_input_counted']
108
109		if 'is_output_counted' not in new_values['data'].keys() or \
110		not isinstance(new_values['data']['is_output_counted'], bool):
111		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
112		description='API.INVALID_IS_OUTPUT_COUNTED_VALUE')
113		is_output_counted = new_values['data']['is_output_counted']
114
115		if 'cost_center_id' not in new_values['data'].keys() or \
116		not isinstance(new_values['data']['cost_center_id'], int) or \
117		new_values['data']['cost_center_id'] <= 0:
118		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
119		description='API.INVALID_COST_CENTER_ID')
120		cost_center_id = new_values['data']['cost_center_id']
121
122		if 'svg_id' in new_values['data'].keys() and \
123		isinstance(new_values['data']['svg_id'], int) and \
124		new_values['data']['svg_id'] > 0:
125		svg_id = new_values['data']['svg_id']
126		else:
127		svg_id = None
128
129		if 'camera_url' in new_values['data'].keys() and \
130		new_values['data']['camera_url'] is not None and \
131		len(str(new_values['data']['camera_url'])) > 0:
132		camera_url = str.strip(new_values['data']['camera_url'])
133		else:
134		camera_url = None
135
136		if 'description' in new_values['data'].keys() and \
137		new_values['data']['description'] is not None and \
138		len(str(new_values['data']['description'])) > 0:
139		description = str.strip(new_values['data']['description'])
140		else:
141		description = None
142
143		cnx = mysql.connector.connect(**config.myems_system_db)
144		cursor = cnx.cursor()
145
146		cursor.execute(" SELECT name "
147		" FROM tbl_combined_equipments "
148		" WHERE name = %s ", (name,))
149		if cursor.fetchone() is not None:
150		cursor.close()
151		cnx.close()
152		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
153		description='API.COMBINED_EQUIPMENT_NAME_IS_ALREADY_IN_USE')
154
155		if cost_center_id is not None:
156		cursor.execute(" SELECT name "
157		" FROM tbl_cost_centers "
158		" WHERE id = %s ",
159		(new_values['data']['cost_center_id'],))
160		row = cursor.fetchone()
161		if row is None:
162		cursor.close()
163		cnx.close()
164		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
165		description='API.COST_CENTER_NOT_FOUND')
166
167		if svg_id is not None:
168		cursor.execute(" SELECT name "
169		" FROM tbl_svgs "
170		" WHERE id = %s ",
171		(svg_id,))
172		row = cursor.fetchone()
173		if row is None:
174		cursor.close()
175		cnx.close()
176		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
177		description='API.SVG_NOT_FOUND')
178
179		add_values = (" INSERT INTO tbl_combined_equipments "
180		" (name, uuid, is_input_counted, is_output_counted, "
181		" cost_center_id, svg_id, camera_url, description) "
182		" VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ")
183		cursor.execute(add_values, (name,
184		str(uuid.uuid4()),
185		is_input_counted,
186		is_output_counted,
187		cost_center_id,
188		svg_id,
189		camera_url,
190		description))
191		new_id = cursor.lastrowid
192		cnx.commit()
193		cursor.close()
194		cnx.close()
195
196		resp.status = falcon.HTTP_201
197		resp.location = '/combinedequipments/' + str(new_id)
198
199
200		class CombinedEquipmentItem:

MyEMS / myems

Code Duplication Length = 188-189 lines in 2 locations

myems-api/core/equipment.py 1 location

myems-api/core/combinedequipment.py 1 location