""" Authentication related API endpoints. """ from typing import Any, Dict from fastapi import Request from ApiLayers.Middleware import MiddlewareModule from Events.Engine.abstract_class import MethodToEvent from Events.base_request_model import EndpointBaseRequestModel, ContextRetrievers from .api_events import ( authentication_login_super_user_event, authentication_select_super_user_event, authentication_check_token_event, authentication_refresh_user_info_event, authentication_change_password_event, authentication_create_password_event, authentication_disconnect_user_event, authentication_logout_user_event, authentication_refresher_token_event, authentication_forgot_password_event, authentication_reset_password_event, authentication_download_avatar_event, ) from .function_handlers import AuthenticationFunctions AuthenticationLoginEventMethods = MethodToEvent( name="AuthenticationLoginEventMethods", events={ authentication_login_super_user_event.key: authentication_login_super_user_event, }, headers=[], errors=[], url="/login", method="POST", summary="Login via domain and access key : [email] | [phone]", description="Login to the system via domain, access key : [email] | [phone]", ) def authentication_login_with_domain_and_creds_endpoint( request: Request, data: EndpointBaseRequestModel ) -> Dict[str, Any]: event_2_catch = AuthenticationLoginEventMethods.retrieve_event( event_function_code=f"{authentication_login_super_user_event.key}" ) data = event_2_catch.REQUEST_VALIDATOR(**data.data) return event_2_catch.endpoint_callable(request=request, data=data) AuthenticationLoginEventMethods.endpoint_callable = ( authentication_login_with_domain_and_creds_endpoint ) AuthenticationSelectEventMethods = MethodToEvent( name="AuthenticationSelectEventMethods", events={ authentication_select_super_user_event.key: authentication_select_super_user_event, }, decorators_list=[MiddlewareModule.auth_required], headers=[], errors=[], url="/select", method="POST", summary="Select company or occupant type", description="Select company or occupant type", ) def authentication_select_company_or_occupant_type( request: Request, data: EndpointBaseRequestModel ) -> Dict[str, Any]: """ Select company or occupant type. """ context_retriever = ContextRetrievers( func=authentication_select_company_or_occupant_type ) function = AuthenticationSelectEventMethods.retrieve_event( event_function_code=f"{authentication_select_super_user_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever data_model = None if context_retriever.token.is_employee: data_model = function.REQUEST_VALIDATOR.get("EmployeeSelection", None)( **data.data ) elif context_retriever.token.is_occupant: data_model = function.REQUEST_VALIDATOR.get("OccupantSelection", None)( **data.data ) return function.endpoint_callable(data=data_model) AuthenticationSelectEventMethods.endpoint_callable = ( authentication_select_company_or_occupant_type ) AuthenticationCheckTokenEventMethods = MethodToEvent( name="AuthenticationCheckTokenEventMethods", events={authentication_check_token_event.key: authentication_check_token_event}, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/check-token", method="POST", summary="Check if token is valid", description="Check if access token is valid for user", ) def authentication_check_token_is_valid(request: Request): context_retriever = ContextRetrievers(func=authentication_check_token_is_valid) function = AuthenticationCheckTokenEventMethods.retrieve_event( event_function_code=f"{authentication_check_token_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable() AuthenticationCheckTokenEventMethods.endpoint_callable = ( authentication_check_token_is_valid ) AuthenticationRefreshEventMethods = MethodToEvent( name="AuthenticationRefreshEventMethods", events={ authentication_refresh_user_info_event.key: authentication_refresh_user_info_event }, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/refresh", method="POST", summary="Refresh user info", description="Refresh user info using access token", ) def authentication_refresh_user_info(request: Request): context_retriever = ContextRetrievers(func=authentication_refresh_user_info) function = AuthenticationRefreshEventMethods.retrieve_event( event_function_code=f"{authentication_refresh_user_info_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable() AuthenticationRefreshEventMethods.endpoint_callable = authentication_refresh_user_info AuthenticationChangePasswordEventMethods = MethodToEvent( name="AuthenticationChangePasswordEventMethods", events={ authentication_change_password_event.key: authentication_change_password_event }, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/change-password", method="POST", summary="Change password", description="Change password with access token", ) def authentication_change_password_event_callable( request: Request, data: EndpointBaseRequestModel ): context_retriever = ContextRetrievers( func=authentication_change_password_event_callable ) function = AuthenticationChangePasswordEventMethods.retrieve_event( event_function_code=f"{authentication_change_password_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable(data=data) AuthenticationChangePasswordEventMethods.endpoint_callable = ( authentication_change_password_event_callable ) AuthenticationCreatePasswordEventMethods = MethodToEvent( name="AuthenticationCreatePasswordEventMethods", events={ authentication_create_password_event.key: authentication_create_password_event }, headers=[], errors=[], url="/create-password", method="POST", summary="Create password", description="Create password with password reset token requested via email", ) def authentication_create_password(request: Request, data: EndpointBaseRequestModel): context_retriever = ContextRetrievers(func=authentication_create_password) function = AuthenticationCreatePasswordEventMethods.retrieve_event( event_function_code=f"{authentication_create_password_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable(data=data) AuthenticationCreatePasswordEventMethods.endpoint_callable = ( authentication_create_password ) AuthenticationDisconnectUserEventMethods = MethodToEvent( name="AuthenticationDisconnectUserEventMethods", events={ authentication_disconnect_user_event.key: authentication_disconnect_user_event }, decorators_list=[MiddlewareModule.auth_required], headers=[], errors=[], url="/disconnect", method="POST", summary="Disconnect all sessions", description="Disconnect all sessions of user in access token", ) def authentication_disconnect_user(request: Request): context_retriever = ContextRetrievers(func=authentication_disconnect_user) function = AuthenticationDisconnectUserEventMethods.retrieve_event( event_function_code=f"{authentication_disconnect_user_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable() AuthenticationDisconnectUserEventMethods.endpoint_callable = ( authentication_disconnect_user ) AuthenticationLogoutEventMethods = MethodToEvent( name="AuthenticationLogoutEventMethods", events={authentication_logout_user_event.key: authentication_logout_user_event}, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/logout", method="POST", summary="Logout user", description="Logout only single session of user which domain is provided", ) def authentication_logout_user(request: Request, data: EndpointBaseRequestModel): context_retriever = ContextRetrievers(func=authentication_logout_user) function = AuthenticationLogoutEventMethods.retrieve_event( event_function_code=f"{authentication_logout_user_event.key}" ) validated_data = function.REQUEST_VALIDATOR(**data.data) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable(data=validated_data) AuthenticationLogoutEventMethods.endpoint_callable = authentication_logout_user AuthenticationRefreshTokenEventMethods = MethodToEvent( name="AuthenticationRefreshTokenEventMethods", events={ authentication_refresher_token_event.key: authentication_refresher_token_event }, headers=[], errors=[], decorators_list=[], url="/refresh-token", method="POST", summary="Refresh token", description="Refresh access token with refresher token", ) def authentication_refresher_token(request: Request, data: EndpointBaseRequestModel): function = AuthenticationRefreshTokenEventMethods.retrieve_event( event_function_code=f"{authentication_refresher_token_event.key}" ) validated_data = function.REQUEST_VALIDATOR(**data.data) return function.endpoint_callable(request=request, data=validated_data) AuthenticationRefreshTokenEventMethods.endpoint_callable = ( authentication_refresher_token ) AuthenticationForgotPasswordEventMethods = MethodToEvent( name="AuthenticationForgotPasswordEventMethods", events={ authentication_forgot_password_event.key: authentication_forgot_password_event }, headers=[], errors=[], url="/forgot-password", method="POST", summary="Request password reset", description="Send an email to user for a valid password reset token", ) def authentication_forgot_password(request: Request, data: EndpointBaseRequestModel): context_retriever = ContextRetrievers(func=authentication_forgot_password) function = AuthenticationForgotPasswordEventMethods.retrieve_event( event_function_code=f"{authentication_forgot_password_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable(data=data) AuthenticationForgotPasswordEventMethods.endpoint_callable = ( authentication_forgot_password ) AuthenticationResetPasswordEventMethods = MethodToEvent( name="AuthenticationResetPasswordEventMethods", events={ authentication_reset_password_event.key: authentication_reset_password_event }, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/reset-password", method="POST", summary="Reset password", description="Reset user password", ) def authentication_reset_password(request: Request, data: EndpointBaseRequestModel): context_retriever = ContextRetrievers(func=authentication_reset_password) function = AuthenticationResetPasswordEventMethods.retrieve_event( event_function_code=f"{authentication_reset_password_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable(data=data.data) AuthenticationResetPasswordEventMethods.endpoint_callable = ( authentication_reset_password ) AuthenticationDownloadAvatarEventMethods = MethodToEvent( name="AuthenticationDownloadAvatarEventMethods", events={ authentication_download_avatar_event.key: authentication_download_avatar_event }, headers=[], errors=[], decorators_list=[MiddlewareModule.auth_required], url="/download-avatar", method="POST", summary="Download avatar", description="Download avatar icon and profile info of user", ) def authentication_download_avatar(request: Request): context_retriever = ContextRetrievers(func=authentication_download_avatar) function = AuthenticationDownloadAvatarEventMethods.retrieve_event( event_function_code=f"{authentication_download_avatar_event.key}" ) AuthenticationFunctions.context_retriever = context_retriever return function.endpoint_callable() AuthenticationDownloadAvatarEventMethods.endpoint_callable = ( authentication_download_avatar )