149 lines
4.9 KiB
Python
149 lines
4.9 KiB
Python
import typing
|
|
from abc import ABC
|
|
from fastapi import status
|
|
from fastapi.exceptions import HTTPException
|
|
from typing import TypeVar, Union, Dict, Any, Optional, Type
|
|
|
|
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
|
|
|
|
TokenType = TypeVar("TokenType", bound=Union[EmployeeTokenObject, OccupantTokenObject])
|
|
|
|
|
|
class ActionsSchema(ABC):
|
|
"""Base class for defining API action schemas.
|
|
|
|
This class handles endpoint registration and validation in the database.
|
|
"""
|
|
|
|
def __init__(self, endpoint: str):
|
|
"""Initialize with an API endpoint path.
|
|
|
|
Args:
|
|
endpoint: The API endpoint path (e.g. "/users/create")
|
|
"""
|
|
self.endpoint = endpoint
|
|
|
|
def retrieve_action_from_endpoint(self) -> Dict[str, Any]:
|
|
"""Retrieve the endpoint registration from the database.
|
|
|
|
Returns:
|
|
Dict containing the endpoint registration data
|
|
|
|
Raises:
|
|
HTTPException: If endpoint is not found in database
|
|
"""
|
|
# Temporarily return a dummy response to skip endpoint restriction checks
|
|
return {
|
|
"endpoint_name": self.endpoint,
|
|
"endpoint_function": "dummy_function",
|
|
"endpoint_method": "GET",
|
|
"endpoint_desc": "Temporary endpoint",
|
|
"endpoint_code": "dummy_code",
|
|
"id": 1,
|
|
"uu_id": "dummy_uuid",
|
|
}
|
|
|
|
|
|
class ActionsSchemaFactory:
|
|
"""Factory class for creating action schemas.
|
|
|
|
This class validates and initializes action schemas for API endpoints.
|
|
"""
|
|
|
|
def __init__(self, action: ActionsSchema):
|
|
"""Initialize with an action schema.
|
|
|
|
Args:
|
|
action: The action schema to initialize
|
|
|
|
Raises:
|
|
HTTPException: If action initialization fails
|
|
"""
|
|
self.action = action
|
|
try:
|
|
self.action_match = self.action.retrieve_action_from_endpoint()
|
|
except HTTPException as e:
|
|
# Re-raise HTTP exceptions as-is
|
|
raise e
|
|
except Exception as e:
|
|
# Log and wrap other exceptions
|
|
print(f"ActionsSchemaFactory Error: {e}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to initialize action schema",
|
|
) from e
|
|
|
|
|
|
class MethodToEvent(ABC, ActionsSchemaFactory):
|
|
"""Base class for mapping methods to API events.
|
|
|
|
This class handles method registration and validation for API events.
|
|
"""
|
|
|
|
action_key: Optional[str] = None
|
|
event_type: Optional[str] = None
|
|
event_description: str = ""
|
|
event_category: str = ""
|
|
|
|
__event_keys__: Dict[str, str] = {}
|
|
__event_validation__: list[Any, list[Any]] = []
|
|
|
|
@classmethod
|
|
def call_event_method(cls, method_uu_id: str, *args: Any, **kwargs: Any) -> Any:
|
|
"""Call an event method by its UUID.
|
|
|
|
Args:
|
|
method_uu_id: UUID of the method to call
|
|
*args: Positional arguments to pass to method
|
|
**kwargs: Keyword arguments to pass to method
|
|
|
|
Returns:
|
|
The result of the called method
|
|
|
|
Raises:
|
|
AttributeError: If method UUID is not found
|
|
"""
|
|
function_name = cls.__event_keys__.get(method_uu_id)
|
|
if not function_name:
|
|
raise AttributeError(f"No method found for UUID: {method_uu_id}")
|
|
return getattr(cls, function_name)(*args, **kwargs)
|
|
|
|
@classmethod
|
|
def ban_token_objects(cls, token: TokenType, ban_list: Type[TokenType]) -> None:
|
|
"""Check if a token type is banned from accessing an event.
|
|
|
|
Args:
|
|
token: The token to check
|
|
ban_list: The token type that is banned
|
|
|
|
Raises:
|
|
HTTPException: If token type matches banned type
|
|
"""
|
|
if isinstance(token, ban_list):
|
|
user_type = (
|
|
"employee" if isinstance(token, EmployeeTokenObject) else "occupant"
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_406_NOT_ACCEPTABLE,
|
|
detail=f"No {user_type} can reach this event. A notification has been sent to admin.",
|
|
)
|
|
|
|
@classmethod
|
|
def retrieve_language_parameters(cls, language: str, function_code: str):
|
|
|
|
event_response_model = dict(cls.__event_validation__).get(function_code)[0]
|
|
event_language_models = list(
|
|
dict(cls.__event_validation__).get(function_code)[1]
|
|
)
|
|
language_models, language_response = {}, {}
|
|
for event_language_model in event_language_models:
|
|
language_models.update({**event_language_model.get(language, "tr")})
|
|
from api_validations.validations_response.building_responses import (
|
|
ListBuildingResponse,
|
|
)
|
|
|
|
for model_field in event_response_model.model_fields:
|
|
if model_field in language_models:
|
|
language_response[model_field] = language_models[model_field]
|
|
return language_response
|