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

95 lines
2.7 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 Users
# List endpoint
SuperUsersListEvent = Event(
name="supers_users_list",
key="341b394f-9f11-4abb-99e7-4b27fa6bf012",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="List events of users endpoint",
)
# Create endpoint
SuperUsersCreateEvent = Event(
name="supers_users_create",
key="4e7e189e-e015-4ff8-902d-60138cbc77a6",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Create events of users endpoint",
)
# Update endpoint
SuperUsersUpdateEvent = Event(
name="supers_users_update",
key="efa4aa4a-d414-4391-91ee-97eb617b7755",
request_validator=None, # TODO: Add request validator
response_validator=None, # TODO: Add response validator
description="Update events of users endpoint",
)
def supers_users_list_callable(list_options: PaginateOnly):
"""
Example callable method
"""
list_options = PaginateOnly(**list_options.model_dump())
with Users.new_session() as db_session:
if list_options.query:
users_list = Users.filter_all(
*Users.convert(list_options.query), db=db_session
)
else:
users_list = Users.filter_all(db=db_session)
pagination = Pagination(data=users_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(
data=users_list,
pagination=pagination,
# response_model="",
).pagination.as_dict
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
SuperUsersListEvent.event_callable = supers_users_list_callable
def supers_users_create_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
SuperUsersCreateEvent.event_callable = supers_users_create_callable
def supers_users_update_callable():
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
SuperUsersUpdateEvent.event_callable = supers_users_update_callable