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())