336 lines
12 KiB
Python
336 lines
12 KiB
Python
"""
|
|
Authentication related API endpoints.
|
|
"""
|
|
|
|
from typing import Union, Any, Dict
|
|
|
|
from ApiLayers.ApiValidations.Custom.wrapper_contexts import AuthContext, EventContext
|
|
from ApiLayers.Middleware import MiddlewareModule, TokenEventMiddleware
|
|
from ApiLayers.ApiValidations.Request import EmployeeSelection, OccupantSelection
|
|
|
|
from Events.Engine.abstract_class import MethodToEvent
|
|
from Events.base_request_model import EndpointBaseRequestModel
|
|
|
|
from .api_events import (
|
|
authentication_login_super_user_event,
|
|
authentication_select_company_or_occupant_type_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 fastapi import Request
|
|
|
|
|
|
# Type aliases for common types
|
|
TokenDictType = Union["EmployeeTokenObject", "OccupantTokenObject"]
|
|
|
|
|
|
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_company_or_occupant_type_super_user_event.key: authentication_select_company_or_occupant_type_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.
|
|
"""
|
|
auth_context = authentication_select_company_or_occupant_type.auth_context
|
|
function = AuthenticationSelectEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_select_company_or_occupant_type_super_user_event.key}"
|
|
)
|
|
function.endpoint_callable.auth_context = auth_context
|
|
return function.endpoint_callable(request=request, data=data)
|
|
|
|
|
|
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):
|
|
function = AuthenticationCheckTokenEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_check_token_event.key}"
|
|
)
|
|
return function.endpoint_callable(request=request)
|
|
|
|
|
|
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):
|
|
token_dict = authentication_refresh_user_info.auth
|
|
function = AuthenticationRefreshEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_refresh_user_info_event.key}"
|
|
)
|
|
return function.endpoint_callable(request=request, token_dict=token_dict)
|
|
|
|
|
|
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):
|
|
token_dict = authentication_change_password_event_callable.auth
|
|
function = AuthenticationChangePasswordEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_change_password_event.key}"
|
|
)
|
|
return function.endpoint_callable(data=data, token_dict=token_dict)
|
|
|
|
|
|
AuthenticationChangePasswordEventMethods.endpoint_callable = authentication_change_password_event_callable
|
|
|
|
AuthenticationCreatePasswordEventMethods = MethodToEvent(
|
|
name="AuthenticationCreatePasswordEventMethods",
|
|
events={
|
|
authentication_create_password_event: 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(data: EndpointBaseRequestModel):
|
|
function = AuthenticationCreatePasswordEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_create_password_event.key}"
|
|
)
|
|
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, data: EndpointBaseRequestModel):
|
|
token_dict = authentication_disconnect_user.auth
|
|
function = AuthenticationDisconnectUserEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_disconnect_user_event.key}"
|
|
)
|
|
return function.endpoint_callable(data=data, token_dict=token_dict)
|
|
|
|
|
|
AuthenticationLogoutEventMethods = MethodToEvent(
|
|
name="AuthenticationLogoutEventMethods",
|
|
events={
|
|
authentication_logout_user_event.key: authentication_logout_user_event
|
|
},
|
|
headers=[],
|
|
errors=[],
|
|
decorators_list=[TokenEventMiddleware.event_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):
|
|
event_context: EventContext = getattr(authentication_logout_user, "event_context", None)
|
|
print('event_context', event_context)
|
|
function = AuthenticationLogoutEventMethods.retrieve_event(event_function_code=f"{event_context.code}")
|
|
function.endpoint_callable.event_context = event_context
|
|
return function.endpoint_callable(request=request, data=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=[MiddlewareModule.auth_required],
|
|
url="/refresh-token",
|
|
method="POST",
|
|
summary="Refresh token",
|
|
description="Refresh access token with refresher token",
|
|
)
|
|
|
|
|
|
def authentication_refresher_token(request: Request, data: EndpointBaseRequestModel):
|
|
auth_context: AuthContext = getattr(authentication_refresher_token, "auth_context", None)
|
|
function = AuthenticationRefreshTokenEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_refresher_token_event.key}"
|
|
)
|
|
function.endpoint_callable.auth_context = auth_context
|
|
return function.endpoint_callable(data=data, request=request)
|
|
|
|
|
|
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):
|
|
token_dict = authentication_forgot_password.auth
|
|
function = AuthenticationForgotPasswordEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_forgot_password_event.key}"
|
|
)
|
|
return function.endpoint_callable(data=data, token_dict=token_dict)
|
|
|
|
|
|
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(data: EndpointBaseRequestModel):
|
|
# token_dict = authentication_reset_password.auth
|
|
function = AuthenticationResetPasswordEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_reset_password_event.key}"
|
|
)
|
|
return function.endpoint_callable(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=[],
|
|
url="/download-avatar",
|
|
method="POST",
|
|
summary="Download avatar",
|
|
description="Download avatar icon and profile info of user",
|
|
)
|
|
|
|
|
|
@MiddlewareModule.auth_required
|
|
def authentication_download_avatar(request: Request):
|
|
token_dict = authentication_download_avatar.auth
|
|
function = AuthenticationDownloadAvatarEventMethods.retrieve_event(
|
|
event_function_code=f"{authentication_download_avatar_event.key}"
|
|
)
|
|
return function.endpoint_callable(token_dict=token_dict)
|
|
|
|
|
|
AuthenticationDownloadAvatarEventMethods.endpoint_callable = authentication_download_avatar
|