376 lines
11 KiB
Python
376 lines
11 KiB
Python
from abc import abstractmethod
|
|
from typing import Any, Dict, List, Optional, Callable
|
|
from uuid import UUID
|
|
|
|
from ApiLayers.AllConfigs.Redis.configs import RedisCategoryKeys
|
|
from Events.base_request_model import STATIC_PATH
|
|
|
|
|
|
class PageComponent:
|
|
|
|
NAME: str
|
|
URL: str
|
|
FETCH_URL: str
|
|
LANGUAGE_MODELS: Dict[str, Any]
|
|
TYPE_COMPONENT: Optional[str] = "Page"
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
url: str,
|
|
# fetch_url: str,
|
|
language_models: Dict[str, Any],
|
|
):
|
|
self.NAME = name
|
|
self.URL = url
|
|
# self.FETCH_URL = fetch_url
|
|
self.LANGUAGE_MODELS = language_models
|
|
|
|
def set_language_models(self, language_models: Dict[str, Any]):
|
|
self.LANGUAGE_MODELS = language_models
|
|
|
|
@property
|
|
def language_models(self):
|
|
return self.LANGUAGE_MODELS
|
|
|
|
def as_dict(self):
|
|
return {
|
|
"name": self.NAME,
|
|
"url": self.URL,
|
|
"language_models": self.LANGUAGE_MODELS,
|
|
# "fetch_url": self.FETCH_URL,
|
|
}
|
|
|
|
|
|
class PageInfo:
|
|
"""
|
|
match_page: {
|
|
"/dashboard?site=AccountCluster": [
|
|
"/accounts/create",
|
|
"/accounts/update",
|
|
"/accounts/list",
|
|
],
|
|
"/update?site=AccountCluster": ["/accounts/update"],
|
|
"/create?site=AccountCluster": ["/accounts/create"],
|
|
},
|
|
"""
|
|
|
|
NAME: str
|
|
PAGE_URL: str
|
|
PAGEINFO: Dict[str, Any]
|
|
URL: str = ""
|
|
ENDPOINTS: Dict[str, Any]
|
|
LANGUAGE_MODELS: Dict[str, Any]
|
|
SUB_COMPONENTS: Optional[list["PageComponent"]] = None
|
|
INSTRUCTIONS: Optional[Dict[str, Any]] = None
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
icon: str,
|
|
url: str,
|
|
endpoints: Dict[str, Any],
|
|
language_models: Dict[str, Any],
|
|
page_info: Optional[Dict[str, Any]] = None,
|
|
sub_components: Optional[list["PageComponent"]] = None,
|
|
instructions: Optional[Dict[str, Any]] = None,
|
|
):
|
|
self.NAME = name
|
|
self.LANGUAGE_MODELS = language_models
|
|
self.ICON = icon
|
|
self.URL = url
|
|
self.SUB_COMPONENTS = sub_components
|
|
self.ENDPOINTS = endpoints
|
|
self.PAGEINFO = page_info
|
|
self.INSTRUCTIONS = instructions
|
|
|
|
@property
|
|
def endpoints(self):
|
|
return self.ENDPOINTS
|
|
|
|
@property
|
|
def sub_components(self):
|
|
return self.SUB_COMPONENTS
|
|
|
|
@property
|
|
def as_dict(self):
|
|
as_dict = {
|
|
"name": self.NAME,
|
|
"icon": self.ICON,
|
|
"url": self.URL,
|
|
"endpoints": self.ENDPOINTS,
|
|
"language_models": self.LANGUAGE_MODELS,
|
|
"page_info": self.PAGEINFO,
|
|
}
|
|
if self.INSTRUCTIONS:
|
|
as_dict["instructions"] = self.INSTRUCTIONS
|
|
if self.SUB_COMPONENTS:
|
|
as_dict["sub_components"] = [i.as_dict() for i in self.SUB_COMPONENTS]
|
|
return as_dict
|
|
|
|
|
|
class Event:
|
|
|
|
KEY_: str # static string uuid.uuid4().__str__()
|
|
RESPONSE_VALIDATOR: Optional[Any]
|
|
REQUEST_VALIDATOR: Optional[Any]
|
|
DESCRIPTION: str
|
|
LANGUAGE_MODELS: list
|
|
STATICS: str
|
|
EXTRA_OPTIONS: Optional[Dict[str, Any]] = None
|
|
endpoint_callable: Any
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
key: str | UUID,
|
|
description: str,
|
|
language_models: list[Dict[str, Dict]],
|
|
statics: str = None,
|
|
request_validator: Optional[Any] = None,
|
|
response_validator: Optional[Any] = None,
|
|
extra_options: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
self.NAME = name
|
|
self.KEY_ = key
|
|
self.REQUEST_VALIDATOR = request_validator
|
|
self.RESPONSE_VALIDATOR = response_validator
|
|
self.STATICS = statics
|
|
self.LANGUAGE_MODELS = language_models
|
|
self.DESCRIPTION = description
|
|
self.EXTRA_OPTIONS = extra_options
|
|
|
|
@property
|
|
def is_static_response(self):
|
|
return bool(self.STATICS)
|
|
|
|
@property
|
|
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.STATICS}"
|
|
]
|
|
)
|
|
if static_response.status:
|
|
return static_response.first
|
|
return None
|
|
|
|
@property
|
|
def static_key(self):
|
|
return self.STATICS
|
|
|
|
@property
|
|
def description(self):
|
|
return f"This is an event of {self.name}. Description: {self.DESCRIPTION}"
|
|
|
|
@property
|
|
def name(self):
|
|
return self.NAME
|
|
|
|
@property
|
|
def key(self) -> str:
|
|
return str(self.KEY_)
|
|
|
|
@abstractmethod
|
|
def endpoint_callable(self, **kwargs) -> Any:
|
|
"""
|
|
Retrieves the endpoint function based on the event key.
|
|
"""
|
|
return self.endpoint_callable(**kwargs)
|
|
|
|
|
|
class MethodToEvent:
|
|
"""
|
|
for all endpoint callable
|
|
def endpoint_callable(request: Request, data: PydanticModel):
|
|
return cls.retrieve_event(event_function_code).retrieve_callable(token_dict, data)
|
|
[Table.__language_model__ | Dict[__language_model__]]
|
|
[Dict[ErrorCode][lang]]
|
|
|
|
"""
|
|
|
|
EVENTS: dict[str, Event]
|
|
HEADER_LANGUAGE_MODELS: list[Dict]
|
|
ERRORS_LANGUAGE_MODELS: Optional[list[Dict]]
|
|
URL: str
|
|
METHOD: str
|
|
SUMMARY: str
|
|
DESCRIPTION: str
|
|
DECORATORS_LIST: Optional[Callable] = []
|
|
EXTRA_OPTIONS: Optional[Dict[str, Any]] = None
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
events: dict[str, Event],
|
|
headers: list[Dict],
|
|
url: str,
|
|
method: str,
|
|
summary: str,
|
|
description: str,
|
|
decorators_list: Optional[List[Callable]] = None,
|
|
errors: Optional[list[Dict]] = None,
|
|
extra_options: Optional[Dict[str, Any]] = None,
|
|
):
|
|
self.EVENTS = events
|
|
self.URL = url
|
|
self.METHOD = method
|
|
self.SUMMARY = summary
|
|
self.NAME = name
|
|
self.DESCRIPTION = description
|
|
self.DECORATORS_LIST = decorators_list
|
|
self.HEADER_LANGUAGE_MODELS = headers
|
|
self.ERRORS_LANGUAGE_MODELS = errors
|
|
self.EXTRA_OPTIONS = extra_options
|
|
|
|
@property
|
|
def name(self):
|
|
return self.NAME
|
|
|
|
def retrieve_all_event_keys(self):
|
|
"""
|
|
Retrieves all event keys from the events list.
|
|
"""
|
|
return [str(event_key) for event_key in self.EVENTS.keys()]
|
|
|
|
def retrieve_event(self, event_function_code: str) -> Event:
|
|
"""
|
|
Retrieves the event object from the events list based on the event function code.
|
|
"""
|
|
if found_event := self.EVENTS.get(event_function_code, None):
|
|
return found_event
|
|
raise ValueError(f"Event with function code {event_function_code} not found")
|
|
|
|
def retrieve_redis_value(self, cluster: "CategoryCluster") -> Dict:
|
|
"""
|
|
Key("METHOD_FUNCTION_CODES:{ClusterToMethod}:MethodEvent:Endpoint") : Value([FUNCTION_CODE, ...])
|
|
"""
|
|
redis_key = f"{RedisCategoryKeys.METHOD_FUNCTION_CODES}:{cluster.name}:{self.name}:{f"{cluster.PREFIX}{self.URL}"}"
|
|
return {redis_key: self.retrieve_all_event_keys()}
|
|
|
|
@staticmethod
|
|
def endpoint_callable(**kwargs):
|
|
"""
|
|
return cls.retrieve_event(event_function_code).retrieve_callable(token_dict, data)
|
|
"""
|
|
raise NotImplementedError("Endpoint callable method is not implemented")
|
|
|
|
|
|
class CategoryCluster:
|
|
|
|
TAGS: list
|
|
PREFIX: str
|
|
PAGEINFO: Optional[Dict["str", PageInfo]]
|
|
DESCRIPTION: str
|
|
ENDPOINTS: dict[str, MethodToEvent] # {"MethodToEvent": MethodToEvent, ...}
|
|
SUBCATEGORY: Optional[List["CategoryCluster"]] # [CategoryCluster, ...]
|
|
MAPPING: Optional[List[Dict[str, Any]]] # [{"key": "value"}, ...]
|
|
INCLUDE_IN_SCHEMA: Optional[bool] = True
|
|
IS_CLIENT: Optional[bool] = False
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
tags: list,
|
|
prefix: str,
|
|
description: str,
|
|
endpoints: dict[str, MethodToEvent],
|
|
sub_category: list,
|
|
mapping: Optional[List[Dict[str, Any]]] = None,
|
|
pageinfo: Optional[Dict["str", PageInfo]] = None,
|
|
include_in_schema: Optional[bool] = True,
|
|
is_client: Optional[bool] = False,
|
|
):
|
|
self.NAME = name
|
|
self.TAGS = tags
|
|
self.PREFIX = prefix
|
|
self.PAGEINFO = pageinfo
|
|
self.DESCRIPTION = description
|
|
self.ENDPOINTS = endpoints or {}
|
|
self.SUBCATEGORY = sub_category or []
|
|
self.INCLUDE_IN_SCHEMA = include_in_schema
|
|
self.MAPPING = mapping
|
|
self.IS_CLIENT = is_client
|
|
|
|
@property
|
|
def is_clickable(self):
|
|
return bool(self.SUBCATEGORY)
|
|
|
|
@property
|
|
def is_client(self):
|
|
return self.IS_CLIENT
|
|
|
|
@property
|
|
def name(self):
|
|
return self.NAME
|
|
|
|
def get_redis_cluster_index_value(self):
|
|
"""
|
|
RedisCategoryKeys.CLUSTER_2_METHOD_EVENT
|
|
Returns the class name and function codes for the class.
|
|
"""
|
|
dict_cluster_2_method, list_endpoints = {}, [
|
|
i.name for i in self.ENDPOINTS.values()
|
|
]
|
|
for endpoint_name in list_endpoints:
|
|
dict_cluster_2_method[endpoint_name] = self.name
|
|
dict_cluster_2_method[self.name] = list_endpoints
|
|
return dict_cluster_2_method
|
|
|
|
def retrieve_all_function_codes(self):
|
|
"""
|
|
Retrieves all function codes by iterating over the events list.
|
|
"""
|
|
all_function_codes = []
|
|
for event_method in self.ENDPOINTS.values():
|
|
all_function_codes.extend(
|
|
[str(event_key) for event_key in event_method.EVENTS.keys()]
|
|
)
|
|
return all_function_codes
|
|
|
|
def retrieve_redis_value(self) -> Dict:
|
|
"""
|
|
Create Redis Key and Value from function codes
|
|
Key(CLUSTER_FUNCTION_CODES:ClusterToMethod) : Value(PAGE_INFO, [FUNCTION_CODE, ...])
|
|
"""
|
|
return {
|
|
f"{RedisCategoryKeys.CLUSTER_FUNCTION_CODES}:{self.name}": self.retrieve_all_function_codes()
|
|
}
|
|
|
|
def retrieve_page_info(self):
|
|
"""
|
|
PAGE_INFO:ClusterToMethod = {"PageInfo": {...}, "subCategory": PAGE_INFO:ClusterToMethod}
|
|
return {"prefix": self.PREFIX, "mapping": self.MAPPING, **page_infos}
|
|
"""
|
|
page_infos = {}
|
|
if isinstance(self.PAGEINFO, dict):
|
|
for page_key, page_info in dict(self.PAGEINFO).items():
|
|
if page_info_dict := getattr(page_info, "as_dict", None):
|
|
page_infos[page_key] = page_info_dict
|
|
return {"prefix": self.PREFIX, "mapping": self.MAPPING, **page_infos}
|
|
if hasattr(self.PAGEINFO, "as_dict"):
|
|
return {"prefix": self.PREFIX, "mapping": self.MAPPING, **self.PAGEINFO.as_dict}
|
|
return
|
|
|
|
|
|
class LanguageModels:
|
|
SITE_URL: str
|
|
COMPONENT: str = "Table"
|
|
PREFIX_URL: str = ""
|
|
PAGE_INFO: dict
|
|
STATIC_PATH: str = STATIC_PATH
|
|
|
|
def as_dict(self):
|
|
return {
|
|
"SITE_URL": f"/{self.STATIC_PATH}{self.SITE_URL}",
|
|
"COMPONENT": self.COMPONENT,
|
|
"PREFIX_URL": self.PREFIX_URL,
|
|
"PAGE_INFO": self.PAGE_INFO,
|
|
}
|
|
|
|
|
|
DefaultClusterName = "site" |