Berkay/wag-managment-api-service-version-5
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

auth response and error codes imp

Browse Source
This commit is contained in:
berkay
2025-01-28 21:35:52 +03:00
parent 8550c2af71
commit a0b1b1bef9
16 changed files with 639 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
Events/AllEvents/authentication/auth/api_events.py
View File

@@ -1,17 +1,20 @@
from Events.Engine.abstract_class import Event
from ApiLayers.Schemas import Users
from ApiLayers.LanguageModels.Request import (
LoginRequestLanguageModel,
SelectRequestLanguageModel,
)
from .models import AuthenticationRequestModels, AuthenticationResponseModels
from .function_handlers import AuthenticationFunctions
from ApiLayers.LanguageModels.Request import LoginRequestLanguageModel
# Auth Login
authentication_login_super_user_event = Event(
name="authentication_login_super_user_event",
key="a5d2d0d1-3e9b-4b0f-8c7d-6d4a4b4c4d4e",
request_validator=AuthenticationRequestModels.LoginSuperUserRequestModel,
language_models=[],
response_validator=AuthenticationResponseModels.LoginSuperUserResponseModel,
language_models=[LoginRequestLanguageModel],
response_validation_static="LOGIN_SUCCESS",
description="Login super user",
)
@@ -26,8 +29,8 @@ authentication_select_super_user_event = Event(
name="authentication_select_super_user_event",
key="a5d2d0d1-3e9b-4b0f-8c7d-6d4a4b4c4d4e",
request_validator=AuthenticationRequestModels.SelectCompanyOrOccupantTypeSuperUserRequestModel,
language_models=[],
response_validator=AuthenticationResponseModels.SelectCompanyOrOccupantTypeSuperUserResponseModel,
language_models=[SelectRequestLanguageModel],
response_validation_static="LOGIN_SELECT",
description="Select company or occupant type super user",
)
@@ -42,7 +45,7 @@ authentication_check_token_event = Event(
name="authentication_check_token_event",
key="b6e3d1e2-4f9c-5c1g-9d8e-7e5f6f5e5d5f",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Check if token is valid",
)
@@ -58,14 +61,14 @@ authentication_refresh_user_info_event = Event(
name="authentication_refresh_user_info_event",
key="c7f4e2f3-5g0d-6d2h-0e9f-8f6g7g6f6e6g",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Refresh user information",
)
authentication_refresh_user_info_event.endpoint_callable = (
AuthenticationFunctions.authentication_refresh_user_info
AuthenticationFunctions.authentication_access_token_user_info
)
@@ -74,7 +77,7 @@ authentication_change_password_event = Event(
name="authentication_change_password_event",
key="d8g5f3g4-6h1e-7e3i-1f0g-9g7h8h7g7f7h",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Change user password",
)
@@ -90,7 +93,7 @@ authentication_create_password_event = Event(
name="authentication_create_password_event",
key="e9h6g4h5-7i2f-8f4j-2g1h-0h8i9i8h8g8i",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Create new password",
)
@@ -106,7 +109,7 @@ authentication_disconnect_user_event = Event(
name="authentication_disconnect_user_event",
key="f0i7h5i6-8j3g-9g5k-3h2i-1i9j0j9i9h9j",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Disconnect all user sessions",
)
@@ -122,7 +125,7 @@ authentication_logout_user_event = Event(
name="authentication_logout_user_event",
key="g1j8i6j7-9k4h-0h6l-4i3j-2j0k1k0j0i0k",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Logout user session",
)
@@ -138,7 +141,7 @@ authentication_refresher_token_event = Event(
name="authentication_refresher_token_event",
key="h2k9j7k8-0l5i-1i7m-5j4k-3k1l2l1k1j1l",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Refresh authentication token",
)
@@ -154,7 +157,7 @@ authentication_forgot_password_event = Event(
name="authentication_forgot_password_event",
key="i3l0k8l9-1m6j-2j8n-6k5l-4l2m3m2l2k2m",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Request password reset",
)
@@ -170,7 +173,7 @@ authentication_reset_password_event = Event(
name="authentication_reset_password_event",
key="j4m1l9m0-2n7k-3k9o-7l6m-5m3n4n3m3l3n",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Reset user password",
)
@@ -186,7 +189,7 @@ authentication_download_avatar_event = Event(
name="authentication_download_avatar_event",
key="k5n2m0n1-3o8l-4l0p-8m7n-6n4o5o4n4m4o",
request_validator=None, # TODO: Add request validator
language_models=[Users.__language_model__],
language_models=[],
# response_validator=None, # TODO: Add response validator
description="Download user avatar and profile info",
)

352
Events/AllEvents/authentication/auth/function_handlers.py
View File

@@ -5,7 +5,11 @@ from ApiLayers.ApiLibrary.common.line_number import get_line_number_for_error
from ApiLayers.ApiServices.Login.user_login_handler import UserLoginModule
from ApiLayers.ApiServices.Token.token_handler import TokenService
from ApiLayers.ApiValidations.Custom.token_objects import CompanyToken, OccupantToken
from ApiLayers.ApiValidations.Response.default_response import EndpointSuccessResponse
from ApiLayers.ApiValidations.Response.default_response import (
EndpointSuccessResponse,
EndpointNotAcceptableResponse,
EndpointBadRequestResponse,
)
from ApiLayers.ErrorHandlers import HTTPExceptionApi
from ApiLayers.Schemas import (
BuildLivingSpace,
@@ -21,6 +25,7 @@ from ApiLayers.Schemas import (
Event2Occupant,
OccupantTypes,
Users,
UsersTokens,
)
from Events.base_request_model import ContextRetrievers, TokenDictType
@@ -149,10 +154,9 @@ class Handlers:
reachable_event_codes = Event2Occupant.get_event_codes(
build_living_space_id=selected_build_living_space.id
)
occupant_type = OccupantTypes.filter_one(
occupant_type = OccupantTypes.filter_one_system(
OccupantTypes.id == selected_build_living_space.occupant_type_id,
db=db,
system=True,
).data
build_part = BuildParts.filter_one(
BuildParts.id == selected_build_living_space.build_parts_id,
@@ -267,199 +271,219 @@ class AuthenticationFunctions:
}
)
@classmethod # Requires not auth context
def authentication_check_token_is_valid(cls, data: Any):
@classmethod # Requires auth context
def authentication_check_token_is_valid(cls):
"""Check if token is valid for user"""
# try:
# if RedisActions.get_object_via_access_key(request=request):
# return ResponseHandler.success("Access Token is valid")
# except HTTPException:
# return ResponseHandler.unauthorized("Access Token is NOT valid")
return
if cls.context_retriever.token:
return EndpointSuccessResponse(
code="TOKEN_VALID", lang=cls.context_retriever.token.lang
).as_dict(data=cls.context_retriever.base)
return {
"completed": False,
"message": "Token is not valid",
}
@classmethod # Requires not auth context
def authentication_refresh_user_info(cls, data: Any):
def authentication_access_token_user_info(cls):
"""Refresh user info using access token"""
# try:
# access_token = request.headers.get(Auth.ACCESS_TOKEN_TAG)
# if not access_token:
# return ResponseHandler.unauthorized()
# found_user = Users.filter_one(Users.uu_id == token_dict.user_uu_id).data
# if not found_user:
# return ResponseHandler.not_found("User not found")
# user_token = UsersTokens.filter_one(
# UsersTokens.domain == found_user.domain_name,
# UsersTokens.user_id == found_user.id,
# UsersTokens.token_type == "RememberMe",
# ).data
# response_data = {
# "access_token": access_token,
# "refresh_token": getattr(user_token, "token", None),
# "user": found_user.get_dict(),
# }
# return ResponseHandler.success(
# "User info refreshed successfully",
# data=response_data,
# )
# except Exception as e:
# return ResponseHandler.error(str(e))
return
if cls.context_retriever.token:
db = Users.new_session()
if found_user := Users.filter_one(Users.id == cls.context_retriever.token.user_id, db=db).data:
return EndpointSuccessResponse(
code="USER_INFO_REFRESHED", lang=cls.context_retriever.token.lang
).as_dict({
"access_token": cls.context_retriever.token, "user": found_user.get_dict(),
})
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={"user": found_user.get_dict()}
)
@classmethod # Requires no auth context
def authentication_change_password(cls, data: Any):
"""Change password with access token"""
# try:
# if not isinstance(token_dict, EmployeeTokenObject):
# return ResponseHandler.unauthorized("Only employees can change password")
# found_user = Users.filter_one(Users.uu_id == token_dict.user_uu_id).data
# if not found_user:
# return ResponseHandler.not_found("User not found")
# if not found_user.check_password(data.old_password):
# return ResponseHandler.unauthorized("Old password is incorrect")
# found_user.set_password(data.new_password)
# return ResponseHandler.success("Password changed successfully")
# except Exception as e:
# return ResponseHandler.error(str(e))
return
if cls.context_retriever.token:
db = Users.new_session()
if found_user := Users.filter_one(Users.id == cls.context_retriever.token.user_id, db=db).data:
found_user.set_password(data.new_password)
return EndpointSuccessResponse(
code="PASSWORD_CHANGED", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict()})
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={"user": found_user.get_dict()}
)
@classmethod # Requires not auth context
def authentication_create_password(cls, data: Any):
"""Create password with password reset token requested via email"""
# if not data.re_password == data.password:
# raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="Password must match")
# if found_user := Users.filter_one(Users.password_token == data.password_token).data:
# found_user.create_password(found_user=found_user, password=data.password)
# found_user.password_token = ""
# found_user.save()
# return ResponseHandler.success("Password is created successfully", data=found_user.get_dict())
# return ResponseHandler.not_found("Record not found")
return
db = Users.new_session()
if not data.re_password == data.password:
return EndpointNotAcceptableResponse(
code="PASSWORD_NOT_MATCH", lang=cls.context_retriever.token.lang
).as_dict(
data={"password": data.password, "re_password": data.re_password}
)
if found_user := Users.filter_one(
Users.password_token == data.password_token, db=db
).data:
found_user.create_password(found_user=found_user, password=data.password)
found_user.password_token = ""
found_user.save()
return EndpointSuccessResponse(
code="CREATED_PASSWORD", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict()})
@classmethod # Requires auth context
def authentication_disconnect_user(cls, data: Any):
def authentication_disconnect_user(cls):
"""Disconnect all sessions of user in access token"""
# found_user = Users.filter_one(Users.uu_id == token_dict.user_uu_id).data
# if not found_user:
# return ResponseHandler.not_found("User not found")
# if already_tokens := RedisActions.get_object_via_user_uu_id(user_id=str(found_user.uu_id)):
# for key, token_user in already_tokens.items():
# RedisActions.delete(key)
# selected_user = Users.filter_one(Users.uu_id == token_user.get("uu_id")).data
# selected_user.remove_refresher_token(domain=data.domain, disconnect=True)
# return ResponseHandler.success("All sessions are disconnected", data=selected_user.get_dict())
# return ResponseHandler.not_found("Invalid data")
return
db = Users.new_session()
found_user = Users.filter_one_system(
Users.id == cls.context_retriever.token.user_id, db=db
).data
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={"user": found_user.get_dict()}
)
registered_tokens = UsersTokens.filter_all(
UsersTokens.user_id == cls.context_retriever.token.id, db=db
)
if registered_tokens.count:
registered_tokens.query.delete()
UsersTokens.save(db=db)
return EndpointSuccessResponse(
code="DISCONNECTED_USER", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict()})
@classmethod # Requires auth context
def authentication_logout_user(cls, data: Any):
"""Logout only single session of user which domain is provided"""
# token_user = None
# if already_tokens := RedisActions.get_object_via_access_key(request=request):
# for key in already_tokens:
# token_user = RedisActions.get_json(key)
# if token_user.get("domain") == data.domain:
# RedisActions.delete(key)
# selected_user = Users.filter_one(Users.uu_id == token_user.get("uu_id")).data
# selected_user.remove_refresher_token(domain=data.domain)
# return ResponseHandler.success("Session is logged out", data=token_user)
# return ResponseHandler.not_found("Logout is not successfully completed")
context_retriever = ContextRetrievers(func=cls.authentication_logout_user)
return context_retriever.base
db = Users.new_session()
found_user = Users.filter_one_system(
Users.id == cls.context_retriever.token.user_id, db=db
).data
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={"user": found_user.get_dict()}
)
registered_tokens = UsersTokens.filter_all_system(
UsersTokens.user_id == cls.context_retriever.token.id,
UsersTokens.domain == cls.context_retriever.token.domain,
db=db,
)
if registered_tokens.count:
registered_tokens.query.delete()
UsersTokens.save(db=db)
return EndpointSuccessResponse(
code="DISCONNECTED_USER", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict()})
@classmethod # Requires not auth context
def authentication_refresher_token(cls, data: Any):
"""Refresh access token with refresher token"""
# token_refresher = UsersTokens.filter_by_one(
# token=data.refresh_token,
# domain=data.domain,
# **UsersTokens.valid_record_dict,
# ).data
# if not token_refresher:
# return ResponseHandler.not_found("Invalid data")
# if found_user := Users.filter_one(Users.id == token_refresher.user_id).data:
# access_key = AuthActions.save_access_token_to_redis(
# request=request, found_user=found_user, domain=data.domain
# )
# found_user.last_agent = request.headers.get("User-Agent", None)
# found_user.last_platform = request.headers.get("Origin", None)
# found_user.last_remote_addr = getattr(request, "remote_addr", None) or request.headers.get("X-Forwarded-For", None)
# found_user.last_seen = str(system_arrow.now())
# response_data = {
# "access_token": access_key,
# "refresh_token": data.refresh_token,
# }
# return ResponseHandler.success("User is logged in successfully via refresher token", data=response_data)
# return ResponseHandler.not_found("Invalid data")
context_retriever = ContextRetrievers(func=cls.authentication_refresher_token)
return context_retriever.base
import arrow
from ApiLayers.ApiServices.Token.token_handler import TokenService
db = UsersTokens.new_session()
token_refresher: UsersTokens = UsersTokens.filter_by_one(
token=data.refresh_token,
domain=data.domain,
db=db,
).data
if not token_refresher:
return EndpointNotAcceptableResponse(
code="REFRESHER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={"refresh_token": data.refresh_token}
)
if found_user := Users.filter_one(Users.id == token_refresher.user_id, db=db).data:
request = cls.context_retriever.request
token_created = TokenService.set_access_token_to_redis(
request=request,
user=found_user,
domain=data.domain,
remember=True,
)
found_user.last_agent = request.headers.get("User-Agent", None)
found_user.last_platform = request.headers.get("Origin", None)
found_user.last_remote_addr = getattr(request, "remote_addr", None) or request.headers.get("X-Forwarded-For", None)
found_user.last_seen = str(arrow.now())
response_data = {
"access_token": token_created.get("access_token"),
"refresh_token": data.refresh_token,
}
return EndpointSuccessResponse(
code="TOKEN_REFRESH", lang=cls.context_retriever.token.lang
).as_dict(data=response_data)
@classmethod # Requires not auth context
def authentication_forgot_password(cls, data: Any):
"""Send an email to user for a valid password reset token"""
# found_user: Users = Users.check_user_exits(access_key=data.access_key, domain=data.domain)
# forgot_key = AuthActions.save_access_token_to_redis(request=request, found_user=found_user, domain=data.domain)
# forgot_link = ApiStatic.forgot_link(forgot_key=forgot_key)
# send_email_completed = send_email(
# subject=f"Dear {found_user.user_tag}, your forgot password link has been sent.",
# receivers=[str(found_user.email)],
# html=change_your_password_template(user_name=found_user.user_tag, forgot_link=forgot_link),
# )
# if not send_email_completed:
# raise HTTPException(status_code=400, detail="Email can not be sent. Try again later")
# found_user.password_token = forgot_key
# found_user.password_token_is_valid = str(system_arrow.shift(days=1))
# found_user.save()
# return ResponseHandler.success("Password is change link is sent to your email or phone", data={})
return
import arrow
from ApiLayers.ApiServices.Token.token_handler import TokenService
from ApiLayers.AllConfigs.Templates.password_templates import change_your_password_template
from Services.Email.send_email import email_sender
from config import ApiStatic
db = Users.new_session()
request = cls.context_retriever.request
found_user: Users = Users.check_user_exits(access_key=data.access_key, domain=data.domain)
forgot_key = TokenService._create_access_token(access=False)
forgot_link = ApiStatic.forgot_link(forgot_key=forgot_key)
send_email_completed = email_sender.send_email(
subject=f"Dear {found_user.user_tag}, your forgot password link has been sent.",
receivers=[str(found_user.email)],
html=change_your_password_template(user_name=found_user.user_tag, forgot_link=forgot_link),
)
if not send_email_completed:
return EndpointBadRequestResponse(
code="EMAIL_NOT_SENT", lang=cls.context_retriever.token.lang
).as_dict(
data={"email": found_user.email}
)
found_user.password_token = forgot_key
found_user.password_token_is_valid = str(arrow.now().shift(days=1))
found_user.save(db=db)
return EndpointSuccessResponse(
code="FORGOT_PASSWORD", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict(), "forgot_link": forgot_link, "token": forgot_key})
@classmethod # Requires not auth context
def authentication_reset_password(cls, data: Any):
"""Reset password with forgot password token"""
# from sqlalchemy import or_
# found_user = Users.query.filter(
# or_(
# Users.email == str(data.access_key).lower(),
# Users.phone_number == str(data.access_key).replace(" ", ""),
# ),
# ).first()
# if not found_user:
# raise HTTPException(
# status_code=status.HTTP_400_BAD_REQUEST,
# detail="Given access key or domain is not matching with the any user record.",
# )
# reset_password_token = found_user.reset_password_token(found_user=found_user)
# send_email_completed = send_email(
# subject=f"Dear {found_user.user_tag}, a password reset request has been received.",
# receivers=[str(found_user.email)],
# html=change_your_password_template(
# user_name=found_user.user_tag,
# forgot_link=ApiStatic.forgot_link(forgot_key=reset_password_token),
# ),
# )
# if not send_email_completed:
# raise found_user.raise_http_exception(status_code=400, message="Email can not be sent. Try again later")
# return ResponseHandler.success("Password change link is sent to your email or phone", data=found_user.get_dict())
return
return cls.context_retriever.base
@classmethod # Requires not auth context
def authentication_download_avatar(cls, data: Any):
"""Download avatar icon and profile info of user"""
# if found_user := Users.filter_one(Users.id == token_dict.user_id).data:
# expired_starts = str(system_arrow.now() - system_arrow.get(str(found_user.expiry_ends)))
# expired_int = (system_arrow.now() - system_arrow.get(str(found_user.expiry_ends))).days
# user_info = {
# "lang": token_dict.lang,
# "full_name": found_user.person.full_name,
# "avatar": found_user.avatar,
# "remember_me": found_user.remember_me,
# "expiry_ends": str(found_user.expiry_ends),
# "expired_str": expired_starts,
# "expired_int": int(expired_int),
# }
# return ResponseHandler.success("Avatar and profile is shared via user credentials", data=user_info)
# return ResponseHandler.not_found("Invalid data")
return
import arrow
db = Users.new_session()
if found_user := Users.filter_one(
Users.id == cls.context_retriever.token.user_id, db=db
).data:
expired_starts = str(arrow.now() - arrow.get(str(found_user.expiry_ends)))
expired_int = arrow.now() - arrow.get(str(found_user.expiry_ends)).days
user_info = {
"lang": cls.context_retriever.token.lang,
"full_name": found_user.person.full_name,
"avatar": found_user.avatar,
"remember_me": found_user.remember_me,
"expiry_ends": str(found_user.expiry_ends),
"expired_str": expired_starts,
"expired_int": int(expired_int),
}
return EndpointSuccessResponse(
code="USER_AVATAR", lang=cls.context_retriever.token.lang
).as_dict(data=user_info)
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict()})

44
Events/AllEvents/authentication/auth/models.py
View File

@@ -1,53 +1,19 @@
from pydantic import BaseModel
from ApiLayers.ApiValidations.Request import (
Login,
EmployeeSelection,
OccupantSelection,
CreatePassword,
ChangePassword,
Forgot,
)
class LoginSuperUserResponseModel(BaseModel):
pass
class SelectCompanyOrOccupantTypeSuperUserRequestModel(BaseModel):
pass
class SelectCompanyOrOccupantTypeSuperUserResponseModel(BaseModel):
pass
class EmployeeSelectionSuperUserRequestModel(BaseModel):
pass
class EmployeeSelectionSuperUserResponseModel(BaseModel):
pass
class OccupantSelectionSuperUserRequestModel(BaseModel):
pass
class OccupantSelectionSuperUserResponseModel(BaseModel):
pass
class AuthenticationRequestModels:
LoginSuperUserRequestModel = Login
SelectCompanyOrOccupantTypeSuperUserRequestModel = {
"EmployeeSelection": EmployeeSelection,
"OccupantSelection": OccupantSelection,
"EmployeeSelection": EmployeeSelection, "OccupantSelection": OccupantSelection,
}
EmployeeSelectionSuperUserRequestModel = EmployeeSelectionSuperUserRequestModel
OccupantSelectionSuperUserRequestModel = OccupantSelectionSuperUserRequestModel
class AuthenticationResponseModels:
LoginSuperUserResponseModel = LoginSuperUserResponseModel
SelectCompanyOrOccupantTypeSuperUserResponseModel = (
SelectCompanyOrOccupantTypeSuperUserResponseModel
)
EmployeeSelectionSuperUserResponseModel = EmployeeSelectionSuperUserResponseModel
OccupantSelectionSuperUserResponseModel = OccupantSelectionSuperUserResponseModel
pass

25
Events/Engine/abstract_class.py
View File

@@ -18,8 +18,7 @@ class PageInfo:
title: Dict[str, Any],
description: Dict[str, Any],
icon: str,
parent: str,
url: str,
parent: str
):
self.NAME = name
self.TITLE = title
@@ -35,6 +34,7 @@ class Event:
REQUEST_VALIDATOR: Optional[Any]
DESCRIPTION: str
LANGUAGE_MODELS: list
RESPONSE_VALIDATOR_STATIC: str
EXTRA_OPTIONS: Optional[Dict[str, Any]] = None
endpoint_callable: Any
@@ -44,6 +44,7 @@ class Event:
key: str | UUID,
description: str,
language_models: list[Dict[str, Dict]],
response_validation_static: str = None,
request_validator: Optional[Any] = None,
response_validator: Optional[Any] = None,
extra_options: Optional[Dict[str, Any]] = None,
@@ -52,17 +53,29 @@ class Event:
self.KEY_ = key
self.REQUEST_VALIDATOR = request_validator
self.RESPONSE_VALIDATOR = response_validator
self.RESPONSE_VALIDATOR_STATIC = response_validation_static
self.LANGUAGE_MODELS = language_models
self.DESCRIPTION = description
self.EXTRA_OPTIONS = extra_options
@property
def request_headers(self):
return self.LANGUAGE_MODELS
def is_static_response(self):
return bool(self.RESPONSE_VALIDATOR_STATIC)
@property
def response_headers(self):
return self.LANGUAGE_MODELS
def static_response(self):
from Services.Redis.Actions.actions import RedisActions
from ApiLayers.AllConfigs.Redis.configs import RedisValidationKeysAction
if self.is_static_response:
static_response = RedisActions.get_json(
list_keys=[
f"{RedisValidationKeysAction.static_response_key}:{self.RESPONSE_VALIDATOR_STATIC}"
]
)
if static_response.status:
return static_response.first
return None
@property
def description(self):