439 lines
13 KiB
Python
439 lines
13 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"],
|
||
},
|
||
|
||
RedisValidation = dict(
|
||
NAME="AccountCluster",
|
||
PREFIX="/accounts",
|
||
URL="/dashboard?site=AccountCluster",
|
||
ICON="Building",
|
||
INFO={
|
||
"en": {
|
||
"page": "Account Records for reaching user all types account information",
|
||
},
|
||
"tr": {
|
||
"page": "Kullanıcı tüm hesap bilgilerine ulaşmak için Hesap Kayıtları",
|
||
},
|
||
},
|
||
SUB_COMPONENTS={
|
||
"/accounts/create": {
|
||
"SITE_URL": "/events/create?site=AccountCluster",
|
||
"COMPONENT": "Link",
|
||
"PREFIX_URL": "/accounts/create",
|
||
"INFO": {
|
||
"en": {
|
||
"page": "Create Account Records",
|
||
"actions": "Actions",
|
||
"table": "Table",
|
||
},
|
||
"tr": {
|
||
"page": "Hesap Kayıdı Oluştur",
|
||
"actions": "Aksiyonlar",
|
||
"table": "Tablo",
|
||
},
|
||
},
|
||
"INSTRUCTIONS": None,
|
||
},
|
||
"/accounts/list": {
|
||
"SITE_URL": "/events/dashboard?site=AccountCluster",
|
||
"COMPONENT": "Table",
|
||
"PREFIX_URL": "/accounts/list",
|
||
"INFO": {
|
||
"en": { "page": "List Account Records" },
|
||
"tr": { "page": "Hesap Kayıtlarını Listele" },
|
||
},
|
||
"INSTRUCTIONS": {
|
||
"headers": {
|
||
"store": True,
|
||
"url": "/validations/header",
|
||
"data": {
|
||
"event_code": "/accounts/list",
|
||
"asked_field": "headers",
|
||
},
|
||
},
|
||
"data": {
|
||
"store": True,
|
||
"url": "/accounts/list",
|
||
"data": {
|
||
"page": 1,
|
||
"limit": 1,
|
||
"order_by": "uu_id",
|
||
"order_type": "desc",
|
||
"query": None,
|
||
},
|
||
},
|
||
},
|
||
},
|
||
},
|
||
)
|
||
"""
|
||
|
||
NAME: str
|
||
PAGE_URL: str
|
||
INFO: 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,
|
||
info: Optional[Dict[str, Any]] = None,
|
||
endpoints: Optional[Dict[str, Any]] = None,
|
||
language_models: Optional[Dict[str, Any]] = None,
|
||
sub_components: Optional[list[Dict[str, Any]]] = 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.INFO = 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,
|
||
"LANGUAGE_MODELS": self.LANGUAGE_MODELS,
|
||
"INFO": self.INFO,
|
||
"SUB_COMPONENTS": 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}
|
||
raise ValueError("PAGE_INFO not found")
|
||
|
||
|
||
class LanguageModels:
|
||
SITE_URL: str
|
||
COMPONENT: str = "Table"
|
||
PREFIX_URL: str = ""
|
||
PAGE_INFO: dict
|
||
STATIC_PATH: str = STATIC_PATH
|
||
ICON: str = ""
|
||
|
||
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,
|
||
"ICON": self.ICON,
|
||
}
|
||
|
||
|
||
DefaultClusterName = "site" |