Jan-23-2019, 06:13 AM
We have created a AWS API gateway for connecting AWS infrastructure. We are having ARC API client for checking the API requests. We have to design a AWS lambda code in such a way that if we type a employee id in the Authorizer request header in the API, it should display in the results whether it is a valid user (if it is, it should return their details) or else it should return a error message. Our AWS lambda is already connected to the API gateway. Please suggest a sample authorizer program for reference for having a user authentication in python, so that I can design accordingly.
"""Rest API for tictof.""" from chalice import Chalice from chalice import Response from util.logger_utility import LoggerUtility from util.response_headers import ResponseHeaders from tic_tof_service import TicTofService from tic_editor import TicEditor from swipes_by_date_service import SwipesByDateService from attendance_service import AttendanceService from mywidget_service import MyWidgetService from pyauthlib import UserInfo, AuthPolicy, HttpMethod, parse_event, raise_401 from my_auth_client import get_client APP = Chalice(app_name='TicTof') @APP.route('/tictof/{employeeid}', methods=['GET']) def get_tic_tof(employeeid): """Get TIC TOF values by employee id.""" # Set log level LoggerUtility.set_level() LoggerUtility.log_debug("Get Tic-Tof request for employee_id: " + str(employeeid)) exception_msg = "There was an error while processing of your request. A mail has been sent to the administrator informing about the same. We sincerely regret the inconvenience caused." try: request = APP.current_request tic_tof_service = TicTofService(str(employeeid), request.query_params) response = tic_tof_service.get_requested_tic_tof() if response is None: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-007", "ReturnMessage": "There are no records available."} return Response(body, status_code=404, headers=ResponseHeaders.get_response_headers()) LoggerUtility.log_info("Get Tic-Tof Response: " + str(response)) return Response(response, status_code=200, headers=ResponseHeaders.get_response_headers()) except AssertionError as assert_error: body = {'Code': '400- BadRequest', 'Message': str(assert_error)} LoggerUtility.log_error(assert_error) return Response(body, status_code=400, headers=ResponseHeaders.get_response_headers()) except KeyError as key_error: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(key_error) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) except Exception as exception: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(exception) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) @APP.route('/tictof/swipes', methods=['PUT']) def edit_tic(): """Edit employee swipes data.""" # Set log level LoggerUtility.set_level() try: request = APP.current_request tic_editor = TicEditor(request.json_body) tic_editor.process_request() LoggerUtility.log_info("Processed edit request successfully!") body = {'Code': '200', 'Message': 'Swipes updated successfully!'} return Response(body, status_code=200, headers=ResponseHeaders.get_response_headers()) except KeyError as key_error: body = {'Code': '400- BadRequest', 'Message': "Malformed request body, " + str(key_error) + " not found!"} return Response(body, status_code=400, headers=ResponseHeaders.get_response_headers()) except TypeError as bad_request: body = {'Code': '400- BadRequest', 'Message': str(bad_request)} return Response(body, status_code=400, headers=ResponseHeaders.get_response_headers()) except Exception as error: body = {'Code': '500- InternalServerError', 'Message': str(error)} response = Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) return response @APP.route('/tictof/swipes/{employeeid}', methods=['GET']) def get_swipes(employeeid): """Get swipes of emp by date.""" # Set log level LoggerUtility.set_level() try: LoggerUtility.log_info('Got request for fetching swipes records of employee: ' + str(employeeid)) request = APP.current_request swipes_by_date_service = SwipesByDateService(employeeid, request.query_params) response = swipes_by_date_service.get_swipe_logs() LoggerUtility.log_info("Processed get request successfully: " + str(response)) return Response(response, status_code=200, headers=ResponseHeaders.get_response_headers()) except KeyError as key_error: body = {'Code': '400- BadRequest', 'Message': "Malformed request body, " + str(key_error) + " not found!"} return Response(body, status_code=400, headers=ResponseHeaders.get_response_headers()) except TypeError as bad_request: body = {'Code': '400- BadRequest', 'Message': str(bad_request)} return Response(body, status_code=400, headers=ResponseHeaders.get_response_headers()) except Exception as error: body = {'Code': '500- InternalServerError', 'Message': str(error)} response = Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) return response @APP.route('/tictof/attendance/{employeeid}', methods=['GET']) def get_attendance(employeeid): """Get attendance of emp by date range.""" # Set log level LoggerUtility.set_level() try: LoggerUtility.log_info('Got request for fetching attendance records of employee: ' + str(employeeid)) exception_msg = "There was an error while processing of your request. A mail has been sent to the administrator informing about the same. We sincerely regret the inconvenience caused." attendance_service = AttendanceService(employeeid) response = attendance_service.get_attendance() LoggerUtility.log_info("Processed get request successfully: " + str(response)) if response is None: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-007", "ReturnMessage": "There are no records available."} return Response(body, status_code=404, headers=ResponseHeaders.get_response_headers()) return Response(response, status_code=200, headers=ResponseHeaders.get_response_headers()) except KeyError as key_error: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(key_error) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) except Exception as exception: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(exception) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) @APP.route('/tictof/mywidget/{employeeid}', methods=['GET']) def get_my_widget(employeeid): """Get attendance of emp by date range.""" # Set log level LoggerUtility.set_level() try: LoggerUtility.log_info('Got request for fetching time analytics detail records of employee: ' + str(employeeid)) exception_msg = "There was an error while processing of your request. A mail has been sent to the administrator informing about the same. We sincerely regret the inconvenience caused." my_widget = MyWidgetService(employeeid) response = my_widget.get_my_widget() LoggerUtility.log_info("Processed get request successfully: " + str(response)) if response is None: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-007", "ReturnMessage": "There are no records available."} return Response(body, status_code=404, headers=ResponseHeaders.get_response_headers()) return Response(response, status_code=200, headers=ResponseHeaders.get_response_headers()) except KeyError as key_error: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(key_error) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers()) except Exception as exception: body = {"ReturnFlag": "F", "ReturnCode": "Error-EP-001", "ReturnMessage": exception_msg} LoggerUtility.log_error(exception) return Response(body, status_code=500, headers=ResponseHeaders.get_response_headers())