import typing from abc import ABC from fastapi.exceptions import HTTPException from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject class ActionsSchema(ABC): def __init__(self, endpoint: str = None): self.endpoint = endpoint def retrieve_action_from_endpoint(self): from databases import EndpointRestriction endpoint_restriction = EndpointRestriction.filter_one( EndpointRestriction.endpoint_name.ilike(f"%{self.endpoint}%") ).get(1) if not endpoint_restriction: raise HTTPException( status_code=404, detail=f"Endpoint {self.endpoint} not found in the database", ) return endpoint_restriction class ActionsSchemaFactory: def __init__(self, action: ActionsSchema): self.action = action try: self.action_match = self.action.retrieve_action_from_endpoint() except Exception as e: err = e self.action_match = None class MethodToEvent(ABC, ActionsSchemaFactory): event_type: str = None event_description: str = "" event_category: str = "" __event_keys__: dict = {} @classmethod def call_event_method(cls, method_uu_id: str, *args, **kwargs): function_name = cls.__event_keys__.get(method_uu_id) return getattr(cls, function_name)(*args, **kwargs) @classmethod def ban_token_objects( cls, token: typing.Union[EmployeeTokenObject, OccupantTokenObject], ban_list: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): from fastapi import status from fastapi.exceptions import HTTPException if token.user_type == ban_list.user_type: if isinstance(token, EmployeeTokenObject): raise HTTPException( status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="No employee can reach this event. An notification is send to admin about event registration", ) if isinstance(token, OccupantTokenObject): raise HTTPException( status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="No occupant can reach this event. An notification is send to admin about event registration", )