prod-wag-backend-automate-s.../ApiServices/ManagementService/Events/application/supers_events.py

149 lines
4.3 KiB
Python

from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
from Schemas import (
Applications,
Application2Employee,
Application2Occupant,
)
# List endpoint
ApplicationListEvent = Event(
name="application_list",
key="b4efda1e-bde7-4659-ab1a-ef74c0fd88b6",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Super Users List applications endpoint",
)
# Create endpoint
ApplicationCreateEvent = Event(
name="application_create",
key="f53ca9aa-5536-4d77-9129-78d67e61db4a",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Super Users Create applications endpoint",
)
# Update endpoint
ApplicationUpdateEvent = Event(
name="application_update",
key="0e9a855e-4e69-44b5-8ac2-825daa32840c",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Super Users Update applications endpoint",
)
#Bind Application to employee
ApplicationBindEmployeeEvent = Event(
name="application_bind_employee",
key="26a96c2d-bca8-41cb-8ac1-f3ca8124434b",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Super Users Application Bind employee endpoint",
)
#Bind Application to occupant
ApplicationBindOccupantEvent = Event(
name="application_bind_occupant",
key="4eaf2bb0-2a42-4d21-ae65-a9259ebee189",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Super Users Application Bind occupant endpoint",
)
def application_list_callable(list_options: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with Applications.new_session() as db_session:
if list_options.query:
applications_list = Applications.filter_all(
*Applications.convert(list_options.query), db=db_session
)
else:
applications_list = Applications.filter_all(db=db_session)
pagination = Pagination(data=applications_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=applications_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
ApplicationListEvent.event_callable = application_list_callable
def application_create_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationCreateEvent.event_callable = application_create_callable
def application_update_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationUpdateEvent.event_callable = application_update_callable
def application_bind_employee_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationBindEmployeeEvent.event_callable = application_bind_employee_callable
def application_bind_occupant_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationBindOccupantEvent.event_callable = application_bind_occupant_callable