122 lines
5.2 KiB
Python
122 lines
5.2 KiB
Python
from typing import Any
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from index import endpoints_index
|
|
from events.events.cluster import EventsEndpointRouterCluster
|
|
|
|
from Validations.defaults.validations import CommonHeaders
|
|
from Validations.response.pagination import PaginateOnly
|
|
from Extensions.Middlewares.token_provider import TokenProvider
|
|
|
|
|
|
# Create API router
|
|
event_endpoint_route = APIRouter(prefix="/events", tags=["Event Actions"])
|
|
|
|
|
|
event_list_available = "EventsListAvailable"
|
|
@event_endpoint_route.post(
|
|
path="/list/available",
|
|
description="List available events endpoint",
|
|
operation_id=endpoints_index[event_list_available],
|
|
)
|
|
def event_list_available_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_list_available)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|
|
|
|
|
|
event_list_appended = "EventsListAppended"
|
|
@event_endpoint_route.post(
|
|
path="/list/appended",
|
|
description="List appended events endpoint",
|
|
operation_id=endpoints_index[event_list_appended],
|
|
)
|
|
def event_list_appended_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_list_appended)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|
|
|
|
|
|
event_register_service = "EventRegisterService"
|
|
@event_endpoint_route.post(
|
|
path="/register/service",
|
|
description="Register service endpoint",
|
|
operation_id=endpoints_index[event_register_service],
|
|
)
|
|
def event_register_service_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_register_service)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|
|
|
|
|
|
event_unregister_service = "EventUnRegisterService"
|
|
@event_endpoint_route.post(
|
|
path="/unregister/service",
|
|
description="Unregister service endpoint",
|
|
operation_id=endpoints_index[event_unregister_service],
|
|
)
|
|
def event_unregister_service_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_unregister_service)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|
|
|
|
|
|
event_bind_employee_extra = "EventBindExtraEmployee"
|
|
@event_endpoint_route.post(
|
|
path="/bind/employee/extra",
|
|
description="Bind employee extra endpoint",
|
|
operation_id=endpoints_index[event_bind_employee_extra],
|
|
)
|
|
def event_bind_employee_extra_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_bind_employee_extra)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|
|
|
|
|
|
event_bind_occupant_extra = "EventBindExtraOccupant"
|
|
@event_endpoint_route.post(
|
|
path="/bind/occupant/extra",
|
|
description="Bind occupant extra endpoint",
|
|
operation_id=endpoints_index[event_bind_occupant_extra],
|
|
)
|
|
def event_bind_occupant_extra_route(
|
|
data: PaginateOnly,
|
|
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
|
|
):
|
|
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
|
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
|
|
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
|
|
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(event_bind_occupant_extra)
|
|
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
|
|
return event_cluster_matched.event_callable(list_options=data)
|