129 lines
4.2 KiB
Python
129 lines
4.2 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__: Dict[str, 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."
|
|
)
|