75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
from fastapi.routing import APIRouter
|
|
from fastapi.requests import Request
|
|
|
|
from api_objects import OccupantTokenObject, EmployeeTokenObject
|
|
from api_validations.validations_request import (
|
|
UpdateEndpointAccessList,
|
|
InsertEndpointAccess,
|
|
)
|
|
|
|
from api_services.redis.auth_actions.token import parse_token_object_to_dict
|
|
from databases import (
|
|
EndpointRestriction,
|
|
Event2Occupant,
|
|
Event2Employee,
|
|
Events,
|
|
)
|
|
|
|
endpoint_restriction_route = APIRouter(prefix="/access", tags=["Endpoint Access"])
|
|
endpoint_restriction_route.include_router(
|
|
endpoint_restriction_route, include_in_schema=True
|
|
)
|
|
|
|
|
|
@endpoint_restriction_route.post(
|
|
path="/endpoint/restriction/create",
|
|
summary="Add extra restriction to endpoints list",
|
|
)
|
|
def endpoint_restriction_create(request: Request, data: InsertEndpointAccess):
|
|
token_dict = parse_token_object_to_dict(request=request)
|
|
return
|
|
|
|
|
|
@endpoint_restriction_route.post(
|
|
path="/endpoint/update", summary="Update extra restriction to endpoints list"
|
|
)
|
|
def endpoint_restriction_update(request: Request, data: UpdateEndpointAccessList):
|
|
token_dict = parse_token_object_to_dict(request=request)
|
|
return
|
|
|
|
|
|
@endpoint_restriction_route.post(
|
|
path="/endpoints/available", summary="List extra restriction to endpoints list"
|
|
)
|
|
def endpoint_restriction_list(request: Request):
|
|
token_dict, records = parse_token_object_to_dict(request=request), []
|
|
if isinstance(token_dict, OccupantTokenObject):
|
|
occupant_events = Event2Occupant.get_event_id_by_build_living_space_id(
|
|
build_living_space_id=token_dict.selected_occupant.living_space_id
|
|
)
|
|
events_list = Events.filter_all(Events.id.in_(occupant_events)).data
|
|
records = EndpointRestriction.filter_all(
|
|
EndpointRestriction.id.in_([event.endpoint_id for event in events_list])
|
|
).data
|
|
elif isinstance(token_dict, EmployeeTokenObject):
|
|
employee_events = Event2Employee.get_event_id_by_employee_id(
|
|
employee_id=token_dict.selected_company.employee_id
|
|
)
|
|
events_list = Events.filter_all(Events.id.in_(employee_events)).data
|
|
records = EndpointRestriction.filter_all(
|
|
EndpointRestriction.id.in_([event.endpoint_id for event in events_list])
|
|
).data
|
|
return dict(
|
|
completed=True,
|
|
message="Available endpoints are listed successfully",
|
|
result=[str(record.endpoint_name) for record in records],
|
|
)
|
|
|
|
|
|
@endpoint_restriction_route.patch(
|
|
path="/endpoint/bind/patch", summary="Patch extra restriction to endpoints list"
|
|
)
|
|
def endpoint_restriction_patch(request: Request):
|
|
token_dict = parse_token_object_to_dict(request=request)
|
|
return
|