Python Forum
Regarding AWS API authorization
Thread Rating:
  • 0 Vote(s) - 0 Average
  • 1
  • 2
  • 3
  • 4
  • 5
Regarding AWS API authorization
#1
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())
Reply


Possibly Related Threads…
Thread Author Replies Views Last Post
  Google Earth Engine Authorization Asmaalashin 0 328 Feb-06-2024, 08:40 PM
Last Post: Asmaalashin

Forum Jump:

User Panel Messages

Announcements
Announcement #1 8/1/2020
Announcement #2 8/2/2020
Announcement #3 8/6/2020