281 lines
9.2 KiB
Python
281 lines
9.2 KiB
Python
from ApiControllers.abstracts.event_clusters import Event
|
|
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
|
|
from Controllers.Postgres.response import EndpointResponse
|
|
from typing import Any
|
|
from Schemas import (
|
|
Events,
|
|
Event2Employee,
|
|
Event2Occupant,
|
|
Event2EmployeeExtra,
|
|
Event2OccupantExtra,
|
|
Service2Events,
|
|
Services,
|
|
)
|
|
|
|
# List available events endpoint
|
|
EventsListAvailableEvent = Event(
|
|
name="event_endpoint_list_available",
|
|
key="d39af512-ec71-4c0f-9b35-e53b0d06d3a4",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users List available events endpoint",
|
|
)
|
|
|
|
# List appended events endpoint
|
|
EventsListAppendedEvent = Event(
|
|
name="event_endpoint_list_appended",
|
|
key="bea77d6a-d99f-468b-9002-b3bda6bb6ad0",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users List appended events endpoint",
|
|
)
|
|
|
|
# Event Register endpoint
|
|
EventRegisterServiceEvent = Event(
|
|
name="event_endpoint_register_service",
|
|
key="e18e7f89-5708-4a15-9258-99b0903ed43d",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Register service endpoint",
|
|
)
|
|
|
|
# Event Unregister endpoint
|
|
EventUnRegisterServiceEvent = Event(
|
|
name="service_endpoint_unregister_service",
|
|
key="4d693774-4857-435b-a63c-c39baebfe916",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Unregister service endpoint",
|
|
)
|
|
|
|
# Bind employee extra endpoint
|
|
EventBindEmployeeExtraEvent = Event(
|
|
name="service_endpoint_bind_employee_extra",
|
|
key="cd452928-4256-4fb4-b81e-0ca41d723616",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Bind service to employee extra endpoint",
|
|
)
|
|
|
|
# Bind occupant extra endpoint
|
|
EventBindOccupantExtraEvent = Event(
|
|
name="service_endpoint_bind_occupant_extra",
|
|
key="cb11a150-8049-45c9-8cf3-d5290ffd2e4a",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Bind service to occupant extra endpoint",
|
|
)
|
|
|
|
|
|
def events_list_available_callable(list_options: PaginateOnly):
|
|
"""
|
|
List available events with pagination and filtering options
|
|
"""
|
|
list_options = PaginateOnly(**list_options.model_dump())
|
|
service_uu_id = list_options.query.get("service_uu_id__ilike", None)
|
|
if not service_uu_id:
|
|
return {
|
|
"message": "MSG0003-PARAM-MISSING",
|
|
"data": list_options.query,
|
|
"completed": False,
|
|
}
|
|
list_options.query.pop("service_uu_id__ilike", None)
|
|
list_options.query.pop("service_uu_id", None)
|
|
with Events.new_session() as db_session:
|
|
service2events = Service2Events.filter_all(
|
|
*Service2Events.convert({"service_uu_id__ilike": service_uu_id}),
|
|
db=db_session,
|
|
)
|
|
already_events = [
|
|
service_to_event.event_id for service_to_event in service2events.data
|
|
]
|
|
if list_options.query:
|
|
events_list = Events.filter_all(
|
|
*Events.convert(list_options.query),
|
|
Events.id.not_in(already_events),
|
|
db=db_session,
|
|
)
|
|
else:
|
|
events_list = Events.filter_all(
|
|
Events.id.not_in(already_events), db=db_session
|
|
)
|
|
pagination = Pagination(data=events_list)
|
|
pagination.change(**list_options.model_dump())
|
|
pagination_result = PaginationResult(data=events_list, pagination=pagination)
|
|
return EndpointResponse(
|
|
message="MSG0003-LIST",
|
|
pagination_result=pagination_result,
|
|
).response
|
|
|
|
|
|
EventsListAvailableEvent.event_callable = events_list_available_callable
|
|
|
|
|
|
def events_list_appended_callable(list_options: PaginateOnly):
|
|
"""
|
|
List appended events with pagination and filtering options
|
|
"""
|
|
list_options = PaginateOnly(**list_options.model_dump())
|
|
service_uu_id = list_options.query.get("service_uu_id__ilike", None)
|
|
if not service_uu_id:
|
|
return {
|
|
"message": "MSG0003-PARAM-MISSING",
|
|
"data": list_options.query,
|
|
"completed": False,
|
|
}
|
|
list_options.query.pop("service_uu_id__ilike", None)
|
|
list_options.query.pop("service_uu_id", None)
|
|
with Events.new_session() as db_session:
|
|
service2events = Service2Events.filter_all(
|
|
*Service2Events.convert({"service_uu_id__ilike": service_uu_id}),
|
|
db=db_session,
|
|
)
|
|
already_events = [
|
|
service_to_event.event_id for service_to_event in service2events.data
|
|
]
|
|
|
|
if list_options.query:
|
|
events_list = Events.filter_all(
|
|
*Events.convert(list_options.query),
|
|
Events.id.in_(already_events),
|
|
db=db_session,
|
|
)
|
|
else:
|
|
events_list = Events.filter_all(
|
|
Events.id.in_(already_events), db=db_session
|
|
)
|
|
pagination = Pagination(data=events_list)
|
|
pagination.change(**list_options.model_dump())
|
|
pagination_result = PaginationResult(data=events_list, pagination=pagination)
|
|
return EndpointResponse(
|
|
message="MSG0003-LIST",
|
|
pagination_result=pagination_result,
|
|
).response
|
|
|
|
|
|
EventsListAppendedEvent.event_callable = events_list_appended_callable
|
|
|
|
|
|
def event_register_service_callable(data: Any):
|
|
"""
|
|
Register event to service
|
|
"""
|
|
with Events.new_session() as db_session:
|
|
event = Events.filter_one_system(
|
|
Events.uu_id == data.event_uu_id, db=db_session
|
|
)
|
|
if not event.data:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service = Services.filter_one_system(
|
|
Services.uu_id == data.service_uu_id, db=db_session
|
|
)
|
|
if not service.data:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_event = Service2Events.find_or_create(
|
|
db=db_session,
|
|
include_args=[
|
|
Service2Events.service_uu_id,
|
|
Service2Events.event_uu_id,
|
|
],
|
|
service_id=service.data.id,
|
|
event_id=event.data.id,
|
|
is_confirmed=True,
|
|
service_uu_id=str(service.data.uu_id),
|
|
event_uu_id=str(event.data.uu_id),
|
|
)
|
|
if not service_to_event.meta_data.created:
|
|
return {
|
|
"message": "MSG0003-ALREADY-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_event.save(db=db_session)
|
|
return {
|
|
"message": "MSG0003-REGISTER",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
EventRegisterServiceEvent.event_callable = event_register_service_callable
|
|
|
|
|
|
def event_unregister_service_callable(data: Any):
|
|
"""
|
|
Unregister event from service
|
|
"""
|
|
with Events.new_session() as db_session:
|
|
event = Events.filter_one_system(
|
|
Events.uu_id == data.event_uu_id, db=db_session
|
|
)
|
|
if not event.data:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service = Services.filter_one_system(
|
|
Services.uu_id == data.service_uu_id, db=db_session
|
|
)
|
|
if not service.data:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_event = Service2Events.filter_one_system(
|
|
Service2Events.service_id == service.data.id,
|
|
Service2Events.event_id == event.data.id,
|
|
db=db_session,
|
|
)
|
|
if not service_to_event.data:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_event.query.delete()
|
|
db_session.commit()
|
|
return {
|
|
"message": "MSG0003-UNREGISTER",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
EventUnRegisterServiceEvent.event_callable = event_unregister_service_callable
|
|
|
|
|
|
def event_bind_employee_extra_callable(data: Any):
|
|
"""
|
|
Bind event to employee extra
|
|
"""
|
|
return {
|
|
"message": "MSG0003-BIND",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
EventBindEmployeeExtraEvent.event_callable = event_bind_employee_extra_callable
|
|
|
|
|
|
def event_bind_occupant_extra_callable(data: Any):
|
|
"""
|
|
Bind event to occupant extra
|
|
"""
|
|
return EndpointResponse(
|
|
message="MSG0003-BIND",
|
|
).response
|
|
|
|
|
|
EventBindOccupantExtraEvent.event_callable = event_bind_occupant_extra_callable
|