single endpoint check

api_configs/configs.py

@@ -33,6 +33,7 @@ class Config:
     ]
     NOT_SECURE_PATHS = [
         "/access/endpoints/available",
+        "/access/endpoint/available"
         "/validations/endpoint",
         "/authentication/avatar",
     ]

api_validations/validations_request/__init__.py

@@ -120,6 +120,7 @@ from .rules import (
     UpdateEndpointAccess,
     UpdateEndpointAccessList,
     InsertEndpointAccess,
+    CheckEndpointAccess,
 )
 from .services import (
     RegisterServices2Employee,
@@ -231,6 +232,7 @@ __all__ = [
     "UpdateEndpointAccess",
     "UpdateEndpointAccessList",
     "InsertEndpointAccess",
+    "CheckEndpointAccess",
     "RegisterServices2Employee",
     "RegisterServices2Occupant",
     "InsertStaff",

api_validations/validations_request/rules.py

@@ -6,6 +6,9 @@ from api_validations.validations_request import (
 )
 
 
+class CheckEndpointAccess(BaseModelRegular):
+    endpoint: str
+
 class InsertEndpointAccess(PydanticBaseModel):
     duty_uu_id: str
     endpoint_restriction_list_uu_ids: list

databases/sql_models/postgres_database.py

@@ -1,4 +1,5 @@
 from api_configs import WagDatabase
+
 # from api_configs import TestDatabase as WagDatabase
 
 from sqlalchemy import create_engine

service_app/routers/rules/router.py

@@ -4,7 +4,7 @@ from fastapi.requests import Request
 from api_objects import OccupantTokenObject, EmployeeTokenObject
 from api_validations.validations_request import (
     UpdateEndpointAccessList,
-    InsertEndpointAccess,
+    InsertEndpointAccess, CheckEndpointAccess,
 )
 
 from api_services.redis.auth_actions.token import parse_token_object_to_dict
@@ -14,6 +14,7 @@ from databases import (
     Event2Employee,
     Events,
 )
+from databases.sql_models.event.event import Services, Service2Events
 
 endpoint_restriction_route = APIRouter(prefix="/access", tags=["Endpoint Access"])
 endpoint_restriction_route.include_router(
@@ -66,6 +67,60 @@ def endpoint_restriction_list(request: Request):
     )
 
 
+@endpoint_restriction_route.post(
+    path="/endpoint/available", summary="List extra restriction to endpoints list"
+)
+def endpoint_restriction_available(request: Request, data: CheckEndpointAccess):
+    token_dict, records = parse_token_object_to_dict(request=request), []
+    endpoint = EndpointRestriction.filter_one(
+        EndpointRestriction.endpoint_name.ilike(f"%{str(data.endpoint)}%")
+    ).data
+    if not endpoint:
+        EndpointRestriction.raise_http_exception(
+            status_code="HTTP_404_NOT_FOUND",
+            error_case="UNAUTHORIZED",
+            message="Only Occupant can see this data",
+            data={},
+        )
+    event = Events.filter_one(Events.id == endpoint.id).data
+    service = Service2Events.filter_one(
+        Service2Events.event_id == event.id,
+    ).data
+    if isinstance(token_dict, OccupantTokenObject):
+        event_occupant = Event2Occupant.filter_one(
+            Event2Occupant.event_service_id == service.id,
+            Event2Occupant.build_living_space_id
+            == token_dict.selected_occupant.living_space_id,
+        ).data
+        if not event_occupant:
+            EndpointRestriction.raise_http_exception(
+                status_code="HTTP_404_NOT_FOUND",
+                error_case="UNAUTHORIZED",
+                message="Only Occupant can see this data",
+                data={},
+            )
+        return dict(
+            completed=True,
+            message="Endpoint is available for this occupant",
+        )
+    elif isinstance(token_dict, EmployeeTokenObject):
+        event_employee = Event2Employee.filter_one(
+            Event2Employee.event_service_id == service.id,
+            Event2Employee.employee_id == token_dict.selected_company.employee_id,
+        ).data
+        if not event_employee:
+            EndpointRestriction.raise_http_exception(
+                status_code="HTTP_404_NOT_FOUND",
+                error_case="UNAUTHORIZED",
+                message="Only Occupant can see this data",
+                data={},
+            )
+        return dict(
+            completed=True,
+            message="Endpoint is available for this occupant",
+        )
+
+
 @endpoint_restriction_route.patch(
     path="/endpoint/bind/patch", summary="Patch extra restriction to endpoints list"
 )

test_environment/test_payment.py

@@ -88,8 +88,9 @@ def decision_book_payment_list():
         Total=[{**item, "type": key} for key, item in dict_books.items()],
     )
 
+
 result = decision_book_payment_list()
-print('result', result)
+print("result", result)
 pprint.pprint(result, indent=2)
 # for key, val in result.items():
 #     print('key', key)