356 lines
13 KiB
Python
356 lines
13 KiB
Python
from typing import Any
|
|
|
|
from api_initializer.event_clusters import Event
|
|
from api_validations.response import (
|
|
PaginateOnly,
|
|
Pagination,
|
|
PaginationResult,
|
|
PostgresResponseSingle,
|
|
PostgresResponse,
|
|
EndpointResponse
|
|
)
|
|
|
|
from schemas import (
|
|
Applications,
|
|
Application2Employee,
|
|
Application2Occupant,
|
|
Service2Application,
|
|
Services,
|
|
)
|
|
|
|
# List all endpoint
|
|
ApplicationListAllEvent = Event(
|
|
name="application_list_all",
|
|
key="1971ce4d-4f59-4aa8-83e2-ca19d7da6d11",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users List all applications endpoint",
|
|
)
|
|
|
|
# List available endpoint
|
|
ApplicationListAvailableEvent = Event(
|
|
name="application_list_available",
|
|
key="d8e733f5-b53a-4c36-9082-12579bf9cc4a",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users List available applications endpoint",
|
|
)
|
|
|
|
# List appended endpoint
|
|
ApplicationListAppendedEvent = Event(
|
|
name="application_list_appended",
|
|
key="ea7bbd58-da09-407c-a630-c324e0272385",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users List appended applications endpoint",
|
|
)
|
|
|
|
# Register application to service endpoint
|
|
ApplicationRegisterServiceEvent = Event(
|
|
name="application_register_service",
|
|
key="47d7cfc8-6004-4442-8357-16ceac5d9d18",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Register application to service endpoint",
|
|
)
|
|
|
|
# Unregister application to service endpoint
|
|
ApplicationUnRegisterServiceEvent = Event(
|
|
name="application_unregister_service",
|
|
key="d228ab26-0b74-440f-8f1f-8f40be5a22f2",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Unregister application to service 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 employee endpoint
|
|
ApplicationBindEmployeeEvent = Event(
|
|
name="application_bind_employee",
|
|
key="948763ee-f221-409e-9a82-8525053505cb",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Bind applications to employee endpoint",
|
|
)
|
|
|
|
# Bind occupant endpoint
|
|
ApplicationBindOccupantEvent = Event(
|
|
name="application_bind_occupant",
|
|
key="03c894a3-b337-4d90-a559-5fcd0dc3e2c5",
|
|
request_validator=None, # TODO: Add request validator
|
|
response_validator=None, # TODO: Add response validator
|
|
description="Super Users Bind applications to occupant endpoint",
|
|
)
|
|
|
|
|
|
def application_list_all_callable(list_options: PaginateOnly):
|
|
list_options = PaginateOnly(**list_options.model_dump())
|
|
with Applications.new_session() as db_session:
|
|
Applications.set_session(db_session)
|
|
if list_options.query:
|
|
applications_list = Applications.query.filter(*Applications.convert(list_options.query))
|
|
else:
|
|
applications_list = Applications.query.filter()
|
|
pagination = Pagination(data=applications_list)
|
|
pagination.change(**list_options.model_dump())
|
|
pagination_result = PaginationResult(data=applications_list, pagination=pagination)
|
|
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
|
|
|
|
|
|
ApplicationListAllEvent.event_callable = application_list_all_callable
|
|
|
|
|
|
def application_list_available_callable(list_options: PaginateOnly):
|
|
"""
|
|
Example callable method
|
|
"""
|
|
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 Applications.new_session() as db_session:
|
|
Service2Application.set_session(db_session)
|
|
service2applications = Service2Application.query.filter(*Service2Application.convert({"service_uu_id__ilike": service_uu_id})).all()
|
|
already_events = [service_to_application.application_id for service_to_application in service2applications]
|
|
if list_options.query:
|
|
applications_list = Applications.query.filter(*Applications.convert(list_options.query), Applications.id.not_in(already_events))
|
|
else:
|
|
applications_list = Applications.query.filter(Applications.id.not_in(already_events))
|
|
pagination = Pagination(data=applications_list)
|
|
pagination.change(**list_options.model_dump())
|
|
pagination_result = PaginationResult(data=applications_list, pagination=pagination)
|
|
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
|
|
|
|
|
|
ApplicationListAvailableEvent.event_callable = application_list_available_callable
|
|
|
|
|
|
def application_list_appended_callable(list_options: PaginateOnly):
|
|
"""
|
|
Example callable method
|
|
"""
|
|
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 Service2Application.new_session() as db_session:
|
|
Service2Application.set_session(db_session)
|
|
Applications.set_session(db_session)
|
|
service2applications = Service2Application.query.filter(*Service2Application.convert({"service_uu_id__ilike": service_uu_id})).all()
|
|
already_events = [service_to_application.application_id for service_to_application in service2applications]
|
|
if list_options.query:
|
|
applications_list = Applications.query.filter(*Applications.convert(list_options.query), Applications.id.in_(already_events))
|
|
else:
|
|
applications_list = Applications.query.filter(Applications.id.in_(already_events))
|
|
pagination = Pagination(data=applications_list)
|
|
pagination.change(**list_options.model_dump())
|
|
pagination_result = PaginationResult(data=applications_list, pagination=pagination)
|
|
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
|
|
|
|
|
|
ApplicationListAppendedEvent.event_callable = application_list_appended_callable
|
|
|
|
|
|
def application_create_callable(data: Any):
|
|
"""
|
|
Create a new application
|
|
"""
|
|
with Applications.new_session() as db_session:
|
|
created_application_dict = data.model_dump()
|
|
Applications.set_session(db_session)
|
|
created_application = Applications.create(**created_application_dict)
|
|
if created_application:
|
|
created_application.save()
|
|
return {
|
|
"completed": True,
|
|
"message": "MSG0001-INSERT",
|
|
"data": created_application,
|
|
}
|
|
return {
|
|
"completed": False,
|
|
"message": "MSG0002-ERROR",
|
|
"data": created_application,
|
|
}
|
|
|
|
|
|
ApplicationCreateEvent.event_callable = application_create_callable
|
|
|
|
|
|
def application_update_callable(data: Any, uu_id: str):
|
|
"""
|
|
Update an existing application
|
|
"""
|
|
with Applications.new_session() as db_session:
|
|
updated_application_dict = data.model_dump(exclude_unset=True, exclude_none=True)
|
|
found_application = Applications.query.filter(Applications.uu_id == uu_id).first()
|
|
if not found_application:
|
|
return {
|
|
"completed": False,
|
|
"message": "MSG0002-FOUND",
|
|
"data": found_application,
|
|
}
|
|
updated_application = found_application.update(**updated_application_dict)
|
|
updated_application.save()
|
|
if updated_application.meta_data.updated:
|
|
return {
|
|
"completed": True,
|
|
"message": "MSG0003-UPDATE",
|
|
"data": updated_application,
|
|
}
|
|
return {
|
|
"completed": False,
|
|
"message": "MSG0003-UPDATE",
|
|
"data": updated_application,
|
|
}
|
|
|
|
|
|
ApplicationUpdateEvent.event_callable = application_update_callable
|
|
|
|
|
|
def application_register_service_callable(data: Any):
|
|
"""
|
|
Register an application to a service
|
|
"""
|
|
with Applications.new_session() as db_session:
|
|
Applications.set_session(db_session)
|
|
event = Applications.query.filter(Applications.uu_id == data.application_uu_id).first()
|
|
if not event:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service = Services.query.filter(Services.uu_id == data.service_uu_id).first()
|
|
if not service:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_application = Service2Application.query.filter(
|
|
Service2Application.service_id == service.data.id,
|
|
Service2Application.application_id == event.data.id,
|
|
).first()
|
|
if service_to_application:
|
|
return {
|
|
"message": "MSG0003-ALREADY-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_application = Service2Application.create(
|
|
service_id=service.data.id,
|
|
service_uu_id=str(service.data.uu_id),
|
|
application_id=event.data.id,
|
|
application_uu_id=str(event.data.uu_id),
|
|
application_code=event.data.application_code,
|
|
site_url=event.data.site_url,
|
|
is_confirmed=True,
|
|
)
|
|
service_to_application.save()
|
|
return {
|
|
"message": "MSG0003-REGISTER",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
ApplicationRegisterServiceEvent.event_callable = application_register_service_callable
|
|
|
|
|
|
def application_unregister_service_callable(data: Any):
|
|
"""
|
|
Unregister an application from a service
|
|
"""
|
|
with Applications.new_session() as db_session:
|
|
Applications.set_session(db_session)
|
|
application = Applications.query.filter(Applications.uu_id == data.application_uu_id).first()
|
|
if not application:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service = Services.query.filter(Services.uu_id == data.service_uu_id).first()
|
|
if not service:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_application = Service2Application.query.filter(
|
|
Service2Application.service_id == service.data.id,
|
|
Service2Application.application_id == application.data.id,
|
|
).first()
|
|
if not service_to_application:
|
|
return {
|
|
"message": "MSG0003-NOT-FOUND",
|
|
"data": data.model_dump(),
|
|
"completed": False,
|
|
}
|
|
service_to_application.query.delete()
|
|
db_session.commit()
|
|
return {
|
|
"message": "MSG0003-UNREGISTER",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
ApplicationUnRegisterServiceEvent.event_callable = application_unregister_service_callable
|
|
|
|
|
|
def application_bind_employee_callable(data: Any):
|
|
"""
|
|
Example callable method
|
|
"""
|
|
return {
|
|
"message": "MSG0003-BIND",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
ApplicationBindEmployeeEvent.event_callable = application_bind_employee_callable
|
|
|
|
|
|
def application_bind_occupant_callable(data: Any):
|
|
"""
|
|
Example callable method
|
|
"""
|
|
return {
|
|
"message": "MSG0003-BIND",
|
|
"data": data.model_dump(),
|
|
"completed": True,
|
|
}
|
|
|
|
|
|
ApplicationBindOccupantEvent.event_callable = application_bind_occupant_callable |