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

instructions and validations tested

Browse Source
This commit is contained in:
berkay
2025-02-06 00:22:15 +03:00
parent b67090087a
commit 378d016d0c
20 changed files with 631 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
Events/AllEvents/validations/validation/api_events.py
View File

@@ -3,9 +3,8 @@ from fastapi import Request
from Events.Engine.abstract_class import Event
from .models import ValidationsPydantic
from .function_handlers import RetrieveValidation
from .models import ValidationsPydantic, ClusterPydantic, PagePydantic
from .function_handlers import RetrieveValidation, RetrievePage
# Validation Event
validation_event = Event(
@@ -41,3 +40,39 @@ def get_menu_by_event_function_code(request: Request, data: Any):
menu_event.endpoint_callable = get_menu_by_event_function_code
# Cluster Event
cluster_event = Event(
name="cluster_event",
key="eed3fe12-cec1-4f35-b43d-62fca0682f73",
request_validator=ClusterPydantic,
language_models=[],
statics=None,
description="Get Left Menu of the user",
)
def get_cluster_by_event_function_code(request: Request, data: Any):
return RetrievePage.retrieve_cluster(data=data)
cluster_event.endpoint_callable = get_cluster_by_event_function_code
# Page Event
page_event = Event(
name="page_event",
key="",
request_validator=PagePydantic,
language_models=[],
statics=None,
description="Get Left Menu of the user",
)
def get_page_by_event_function_code(request: Request, data: Any):
return RetrievePage.retrieve_page(data=data)
page_event.endpoint_callable = get_page_by_event_function_code

4
Events/AllEvents/validations/validation/cluster.py
View File

@@ -3,6 +3,8 @@ from Events.Engine.abstract_class import CategoryCluster
from .validation import (
ValidationEventMethods,
MenuEventMethods,
ClusterEventMethods,
PageEventMethods,
)
@@ -14,6 +16,8 @@ ValidationsCluster = CategoryCluster(
endpoints={
"ValidationEventMethods": ValidationEventMethods,
"MenuEventMethods": MenuEventMethods,
"ClusterEventMethods": ClusterEventMethods,
"PageEventMethods": PageEventMethods,
},
include_in_schema=True,
sub_category=[],

91
Events/AllEvents/validations/validation/function_handlers.py
View File

@@ -121,3 +121,94 @@ class RetrieveValidation(BaseRouteModel):
return RedisHeaderRetrieve(**validate_dict).header
elif data.asked_field == "validation":
return RedisValidationRetrieve(**validate_dict).validation
class RetrievePage(BaseRouteModel):
@staticmethod
def get_site_cluster(page_name: str):
"""
/dashboard?site=clusterName retrieves clusterName
"""
if page_name:
return page_name.split("?")[1].split("=")[1]
raise ValueError("Page name not found")
@classmethod
def retrieve_cluster(cls, data: Any):
"""
Retrieve cluster by event function code
"""
reachable_codes = []
if cls.context_retriever.token.is_employee:
reachable_codes = (
cls.context_retriever.token.selected_company.reachable_event_codes
)
elif cls.context_retriever.token.is_occupant:
reachable_codes = (
cls.context_retriever.token.selected_occupant.reachable_event_codes
)
validate_dict = dict(url=data.url, reachable_code=reachable_codes)
print("validate_dict", validate_dict)
cluster_name = data.get("name")
print("cluster_name", cluster_name)
raise NotImplementedError("Cluster not found")
@classmethod
def retrieve_page(cls, data: Any):
"""
Retrieve page by event function code
"""
from Events.Engine import CategoryCluster
from Events.JustEvents.events_file import retrieve_cluster_by_name
reachable_codes = []
if cls.context_retriever.token.is_employee:
reachable_codes = (
cls.context_retriever.token.selected_company.reachable_event_codes
)
elif cls.context_retriever.token.is_occupant:
reachable_codes = (
cls.context_retriever.token.selected_occupant.reachable_event_codes
)
cluster_from_all_events = cls.get_site_cluster(page_name=data.page)
if not cluster_from_all_events:
raise ValueError(f"Cluster not found : {data.page}")
cluster_name: CategoryCluster = retrieve_cluster_by_name(cluster_from_all_events)
if not cluster_name:
raise ValueError("Cluster not found")
page_info = cluster_name.retrieve_page_info().get(data.page, None)
if not page_info:
raise ValueError("Page not found")
endpoints: dict = dict(page_info).get("endpoints", {})
if not endpoints:
raise ValueError("Endpoints not found")
new_page_info_dict = dict(
name=cluster_name.name,
prefix=cluster_name.PREFIX,
url=dict(page_info).get("url", None),
icon=dict(page_info).get("icon", None),
page_info=dict(page_info).get("page_info", None),
endpoints={},
language_models={},
instructions={},
)
for key, event_codes in dict(endpoints).items():
if set(event_codes) & set(reachable_codes):
language_models = dict(page_info).get("language_models", {})
instructions = dict(page_info).get("instructions", {})
new_page_info_dict["endpoints"][key] = True
if language_models.get(key, None):
if key in language_models[key].keys(): # key has sub key blocks inside lang model
for key_model, val_model in dict(language_models[key]).items():
if key_model in new_page_info_dict["endpoints"].keys():
new_page_info_dict["language_models"][key_model] = language_models[key][key_model]
else:
new_page_info_dict["language_models"][key] = language_models[key]
if instructions.get(key, None):
new_page_info_dict["instructions"][key] = instructions.get(key, None)
return new_page_info_dict

8
Events/AllEvents/validations/validation/models.py
View File

@@ -9,3 +9,11 @@ from pydantic import BaseModel
class ValidationsPydantic(BaseModel):
event_code: str
asked_field: Optional[str] = "all"
class ClusterPydantic(BaseModel):
name: str
class PagePydantic(BaseModel):
page: str

62
Events/AllEvents/validations/validation/validation.py
View File

@@ -9,8 +9,8 @@ from Events.Engine.abstract_class import MethodToEvent
from Events.base_request_model import EndpointBaseRequestModel, ContextRetrievers
from ApiLayers.Middleware.auth_middleware import MiddlewareModule
from .api_events import validation_event, menu_event
from .function_handlers import RetrieveValidation
from .api_events import validation_event, menu_event, cluster_event, page_event
from .function_handlers import RetrieveValidation, RetrievePage
ValidationEventMethods = MethodToEvent(
@@ -54,12 +54,66 @@ MenuEventMethods = MethodToEvent(
def menu_endpoint(request: Request, data: EndpointBaseRequestModel) -> Dict[str, Any]:
function = ValidationEventMethods.retrieve_event(
function = MenuEventMethods.retrieve_event(
event_function_code=f"{menu_event.key}"
)
data = function.REQUEST_VALIDATOR(**data.data)
RetrieveValidation.context_retriever = ContextRetrievers(func=validations_endpoint)
RetrieveValidation.context_retriever = ContextRetrievers(func=menu_endpoint)
return function.endpoint_callable(request=request, data=data)
MenuEventMethods.endpoint_callable = menu_endpoint
# Cluster Event
ClusterEventMethods = MethodToEvent(
name="ClusterEventMethods",
events={cluster_event.key: cluster_event},
headers=[],
errors=[],
url="/cluster",
method="POST",
decorators_list=[MiddlewareModule.auth_required],
summary="Get Left Menu of the user",
description="Get Left Menu of the user",
)
def cluster_endpoint(
request: Request, data: EndpointBaseRequestModel
) -> Dict[str, Any]:
function = ClusterEventMethods.retrieve_event(
event_function_code=f"{cluster_event.key}"
)
data = function.REQUEST_VALIDATOR(**data.data)
RetrievePage.context_retriever = ContextRetrievers(func=cluster_endpoint)
return function.endpoint_callable(request=request, data=data)
ClusterEventMethods.endpoint_callable = cluster_endpoint
# Page Event
PageEventMethods = MethodToEvent(
name="PageEventMethods",
events={page_event.key: page_event},
headers=[],
errors=[],
url="/page",
method="POST",
decorators_list=[MiddlewareModule.auth_required],
summary="Get Left Menu of the user",
description="Get Left Menu of the user",
)
def page_endpoint(request: Request, data: EndpointBaseRequestModel) -> Dict[str, Any]:
function = PageEventMethods.retrieve_event(
event_function_code=f"{page_event.key}"
)
data = function.REQUEST_VALIDATOR(**data.data)
RetrievePage.context_retriever = ContextRetrievers(func=page_endpoint)
return function.endpoint_callable(request=request, data=data)
PageEventMethods.endpoint_callable = page_endpoint