evyos-center-server/prod-wag-backend-automate-services
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

appender events and applications are updated

Browse Source
This commit is contained in:
Berkay
2025-05-06 12:03:58 +03:00
parent dd707b2463
commit e0ae1ee80a
53 changed files with 2358 additions and 829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
ApiServices/DealerService/app.py
View File

@@ -1,12 +1,12 @@
from Controllers.Postgres.database import get_db
from Schemas import Users, Employees, BuildLivingSpace
from Schemas import Users, Employees, BuildLivingSpace, Services
from init_service_to_events import init_service_to_event_matches_for_super_user
from init_applications import (
init_applications_for_super_user,
init_applications_for_general_manager,
init_applications_for_build_manager,
init_applications_for_owner,
init_applications_for_tenant,
# init_applications_for_general_manager,
# init_applications_for_build_manager,
# init_applications_for_owner,
# init_applications_for_tenant,
)
@@ -16,62 +16,59 @@ if __name__ == "__main__":
Set Events to service | Set Service to employee
"""
with get_db() as db_session:
if super_man := Users.filter_one(
Users.email == "karatay.berkay.sup@evyos.com.tr", db=db_session
).data:
super_employee = Employees.filter_one(
Employees.people_id == super_man.person_id, db=db_session
).data
service_super_user = Services.filter_one(Services.service_code == "SRE-SUE", db=db_session).data
user_super_user = Users.filter_one(Users.email == "karatay.berkay.sup@evyos.com.tr", db=db_session).data
employee_super_user = Employees.filter_one(Employees.people_id == user_super_user.person_id, db=db_session).data
if service_super_user and user_super_user and employee_super_user:
init_service_to_event_matches_for_super_user(
super_user=super_employee, db_session=db_session
service_match=service_super_user, employee_match=employee_super_user, db_session=db_session
)
init_applications_for_super_user(
super_user=super_employee, db_session=db_session
service_match=service_super_user, employee_match=employee_super_user, db_session=db_session
)
with get_db() as db_session:
print("Createing GM")
if gen_man := Users.filter_one(
Users.email == "example.general@evyos.com.tr", db=db_session
).data:
gen_man_employee = Employees.filter_one(
Employees.people_id == gen_man.person_id, db=db_session
).data
print("General Manager : ", gen_man_employee)
init_applications_for_general_manager(
super_user=gen_man_employee, db_session=db_session
)
# with get_db() as db_session:
# print("Createing GM")
# if gen_man := Users.filter_one(
# Users.email == "example.general@evyos.com.tr", db=db_session
# ).data:
# gen_man_employee = Employees.filter_one(
# Employees.people_id == gen_man.person_id, db=db_session
# ).data
# print("General Manager : ", gen_man_employee)
# init_applications_for_general_manager(
# super_user=gen_man_employee, db_session=db_session
# )
with get_db() as db_session:
if build_man := Users.filter_one(
Users.email == "example.build.manager@gmail.com", db=db_session
).data:
build_man_employee = BuildLivingSpace.filter_one(
BuildLivingSpace.person_id == build_man.person_id, db=db_session
).data
init_applications_for_build_manager(
super_user=build_man_employee, db_session=db_session
)
# with get_db() as db_session:
# if build_man := Users.filter_one(
# Users.email == "example.build.manager@gmail.com", db=db_session
# ).data:
# build_man_employee = BuildLivingSpace.filter_one(
# BuildLivingSpace.person_id == build_man.person_id, db=db_session
# ).data
# init_applications_for_build_manager(
# super_user=build_man_employee, db_session=db_session
# )
with get_db() as db_session:
if own_flt := Users.filter_one(
Users.email == "example.owner@gmail.com", db=db_session
).data:
own_flt_employee = BuildLivingSpace.filter_one(
BuildLivingSpace.person_id == own_flt.person_id, db=db_session
).data
init_applications_for_owner(
super_user=own_flt_employee, db_session=db_session
)
# with get_db() as db_session:
# if own_flt := Users.filter_one(
# Users.email == "example.owner@gmail.com", db=db_session
# ).data:
# own_flt_employee = BuildLivingSpace.filter_one(
# BuildLivingSpace.person_id == own_flt.person_id, db=db_session
# ).data
# init_applications_for_owner(
# super_user=own_flt_employee, db_session=db_session
# )
with get_db() as db_session:
if ten_flt := Users.filter_one(
Users.email == "example.tenant@gmail.com", db=db_session
).data:
ten_flt_employee = BuildLivingSpace.filter_one(
BuildLivingSpace.person_id == ten_flt.person_id, db=db_session
).data
init_applications_for_tenant(
super_user=ten_flt_employee, db_session=db_session
)
# with get_db() as db_session:
# if ten_flt := Users.filter_one(
# Users.email == "example.tenant@gmail.com", db=db_session
# ).data:
# ten_flt_employee = BuildLivingSpace.filter_one(
# BuildLivingSpace.person_id == ten_flt.person_id, db=db_session
# ).data
# init_applications_for_tenant(
# super_user=ten_flt_employee, db_session=db_session
# )

375
ApiServices/DealerService/init_applications.py
View File

@@ -4,222 +4,193 @@ from Schemas import (
Application2Occupant,
Employees,
BuildLivingSpace,
Services,
Service2Application,
)
def init_applications_for_super_user(super_user: Employees, db_session=None) -> None:
list_of_created_apps = [
dict(
name="Dashboard1",
application_code="app000001",
site_url="/dashboard",
application_type="info",
description="Dashboard Page",
),
dict(
name="Individual",
application_code="app000003",
site_url="/individual",
application_type="Dash",
description="Individual Page for people",
),
dict(
name="User",
application_code="app000004",
site_url="/user",
application_type="Dash",
description="Individual Page for user",
),
dict(
name="Build",
application_code="app000005",
site_url="/build",
application_type="Dash",
description="Individual Page for build",
),
dict(
name="BuildParts",
application_code="app000006",
site_url="/build/parts",
application_type="Dash",
description="Individual Page for build parts",
),
dict(
name="BuildArea",
application_code="app000007",
site_url="/build/area",
application_type="Dash",
description="Individual Page for build area",
),
dict(
name="ManagementAccounting",
application_code="app000008",
site_url="/management/accounting",
application_type="Dash",
description="Individual Page for management accounting",
),
dict(
name="ManagementBudget",
application_code="app000009",
site_url="/management/budget",
application_type="Dash",
description="Individual Page for management accounting2",
),
dict(
name="ManagementMeetingClose",
application_code="app000010",
site_url="/annual/meeting/close",
application_type="Dash",
description="Individual Page for management accounting3",
),
dict(
name="EmergencyMeeting",
application_code="app000011",
site_url="/emergency/meeting",
application_type="Dash",
description="Individual Page for management accounting4",
),
dict(
name="EmergencyMeetingClose",
application_code="app000012",
site_url="/emergency/meeting/close",
application_type="Dash",
description="Individual Page for management accounting5",
),
dict(
name="MeetingParticipation",
application_code="app000013",
site_url="/meeting/participation",
application_type="Dash",
description="Individual Page for management accounting6",
),
]
for list_of_created_app in list_of_created_apps:
created_page = Applications.find_or_create(
**list_of_created_app,
def init_applications_for_super_user(service_match: Services, employee_match: Employees, db_session=None) -> None:
list_of_all_events = Applications.filter_all(db=db_session).data
Service2Application.filter_all(db=db_session).query.delete()
Service2Application.save(db=db_session)
for list_of_event_code in list_of_all_events:
service_to_event_found = Service2Application.filter_one_system(
Service2Application.application_id == list_of_event_code.id,
Service2Application.service_id == service_match.id,
db=db_session,
)
if service_to_event_found.data:
service_to_event_found.destroy(db=db_session)
print(
f"UUID: {service_to_event_found.uu_id} application is deleted from {service_match.service_description}"
)
employee_added_application = Service2Application.find_or_create(
service_id=service_match.id,
service_uu_id=str(service_match.uu_id),
application_id=list_of_event_code.id,
application_uu_id=str(list_of_event_code.uu_id),
application_code=list_of_event_code.application_code,
site_url=list_of_event_code.site_url,
is_confirmed=True,
)
if created_page.meta_data.created:
created_page.save(db=db_session)
application_employee_created = Application2Employee.find_or_create(
employee_id=super_user.id,
employee_uu_id=str(super_user.uu_id),
site_url=created_page.site_url,
application_code=created_page.application_code,
application_id=created_page.id,
application_uu_id=str(created_page.uu_id),
is_confirmed=True,
active=True,
db=db_session,
)
if application_employee_created.meta_data.created:
application_employee_created.save(db=db_session)
if employee_added_application.meta_data.created:
employee_added_application.save(db=db_session)
print(
f"UUID: {employee_added_application.uu_id} application is saved to {service_match.service_description}"
)
def init_applications_for_general_manager(
super_user: Employees, db_session=None
) -> None:
list_of_created_apps = [
dict(
name="Dashboard1",
application_code="app000001",
site_url="/dashboard",
application_type="info",
description="Dashboard Page",
),
dict(
name="ManagementAccounting",
application_code="app000008",
site_url="/management/accounting",
application_type="Dash",
description="Individual Page for management accounting",
),
]
for list_of_created_app in list_of_created_apps:
created_page = Applications.find_or_create(
**list_of_created_app,
db=db_session,
employee_added_service = Application2Employee.find_or_create(
service_id=service_match.id,
service_uu_id=str(service_match.uu_id),
employee_id=employee_match.id,
employee_uu_id=str(employee_match.uu_id),
is_confirmed=True,
db=db_session,
)
if employee_added_service.meta_data.created:
employee_added_service.save(db=db_session)
print(
f"UUID: {employee_added_service.uu_id} service is saved to {employee_match.uu_id}"
)
print("Application : ", created_page)
if created_page.meta_data.created:
created_page.save(db=db_session)
application_employee_created = Application2Employee.find_or_create(
employee_id=super_user.id,
employee_uu_id=str(super_user.uu_id),
site_url=created_page.site_url,
application_code=created_page.application_code,
application_id=created_page.id,
application_uu_id=str(created_page.uu_id),
is_confirmed=True,
db=db_session,
)
print("Application Employee : ", application_employee_created)
if application_employee_created.meta_data.created:
application_employee_created.save(db=db_session)
def init_applications_for_build_manager(
super_user: BuildLivingSpace, db_session=None
) -> None:
list_of_created_apps = []
# def init_applications_for_general_manager(
# super_user: Employees, db_session=None
# ) -> None:
# list_of_created_apps = Applications.filter_all_system(db=db_session).data
# if not list_of_created_apps:
# raise Exception("No applications found")
# for list_of_created_app in list_of_created_apps:
# application_employee_created = Application2Employee.find_or_create(
# employee_id=super_user.id,
# employee_uu_id=str(super_user.uu_id),
# site_url=list_of_created_app.site_url,
# application_code=list_of_created_app.application_code,
# application_id=list_of_created_app.id,
# application_uu_id=str(list_of_created_app.uu_id),
# is_confirmed=True,
# db=db_session,
# )
# if application_employee_created.meta_data.created:
# application_employee_created.save(db=db_session)
def init_applications_for_owner(super_user: BuildLivingSpace, db_session=None) -> None:
pass
# def init_applications_for_build_manager(
# super_user: BuildLivingSpace, db_session=None
# ) -> None:
# list_of_created_apps = Applications.filter_all_system(db=db_session).data
# if not list_of_created_apps:
# raise Exception("No applications found")
# for list_of_created_app in list_of_created_apps:
# application_employee_created = Application2Employee.find_or_create(
# employee_id=super_user.id,
# employee_uu_id=str(super_user.uu_id),
# site_url=list_of_created_app.site_url,
# application_code=list_of_created_app.application_code,
# application_id=list_of_created_app.id,
# application_uu_id=str(list_of_created_app.uu_id),
# is_confirmed=True,
# db=db_session,
# )
# if application_employee_created.meta_data.created:
# application_employee_created.save(db=db_session)
def init_applications_for_tenant(super_user: BuildLivingSpace, db_session=None) -> None:
list_of_created_apps = [
dict(
name="Dashboard1",
application_code="app000001",
site_url="/dashboard",
application_type="info",
description="Dashboard Page",
),
dict(
name="TenantSendMessageToBuildManager",
application_code="app000022",
site_url="/tenant/messageToBM",
application_type="Dash",
description="Individual Page for tenant send message to build manager",
),
dict(
name="TenantSendMessageToOwner",
application_code="app000018",
site_url="/tenant/messageToOwner",
application_type="Dash",
description="Individual Page for tenant send message to owner",
),
dict(
name="TenantAccountView",
application_code="app000019",
site_url="/tenant/accounting",
application_type="Dash",
description="Individual Page for tenant account view",
),
]
# def init_applications_for_owner(super_user: BuildLivingSpace, db_session=None) -> None:
# list_of_created_apps = Applications.filter_all_system(db=db_session).data
# if not list_of_created_apps:
# raise Exception("No applications found")
for list_of_created_app in list_of_created_apps:
created_page = Applications.find_or_create(
**list_of_created_app,
db=db_session,
is_confirmed=True,
)
if created_page.meta_data.created:
created_page.save(db=db_session)
# for list_of_created_app in list_of_created_apps:
# application_employee_created = Application2Employee.find_or_create(
# employee_id=super_user.id,
# employee_uu_id=str(super_user.uu_id),
# site_url=list_of_created_app.site_url,
# application_code=list_of_created_app.application_code,
# application_id=list_of_created_app.id,
# application_uu_id=str(list_of_created_app.uu_id),
# is_confirmed=True,
# db=db_session,
# )
# if application_employee_created.meta_data.created:
# application_employee_created.save(db=db_session)
application_occupant_created = Application2Occupant.find_or_create(
build_living_space_id=super_user.id,
build_living_space_uu_id=str(super_user.uu_id),
site_url=created_page.site_url,
application_code=created_page.application_code,
application_id=created_page.id,
application_uu_id=str(created_page.uu_id),
is_confirmed=True,
db=db_session,
)
if application_occupant_created.meta_data.created:
application_occupant_created.save(db=db_session)
# def init_applications_for_tenant(super_user: BuildLivingSpace, db_session=None) -> None:
# list_of_created_apps = Applications.filter_all_system(db=db_session).data
# if not list_of_created_apps:
# raise Exception("No applications found")
# for list_of_created_app in list_of_created_apps:
# application_employee_created = Application2Employee.find_or_create(
# employee_id=super_user.id,
# employee_uu_id=str(super_user.uu_id),
# site_url=list_of_created_app.site_url,
# application_code=list_of_created_app.application_code,
# application_id=list_of_created_app.id,
# application_uu_id=str(list_of_created_app.uu_id),
# is_confirmed=True,
# db=db_session,
# )
# if application_employee_created.meta_data.created:
# application_employee_created.save(db=db_session)
# list_of_created_apps = [
# dict(
# name="Dashboard1",
# application_code="app000001",
# site_url="/dashboard",
# application_type="info",
# description="Dashboard Page",
# ),
# dict(
# name="TenantSendMessageToBuildManager",
# application_code="app000022",
# site_url="/tenant/messageToBM",
# application_type="Dash",
# description="Individual Page for tenant send message to build manager",
# ),
# dict(
# name="TenantSendMessageToOwner",
# application_code="app000018",
# site_url="/tenant/messageToOwner",
# application_type="Dash",
# description="Individual Page for tenant send message to owner",
# ),
# dict(
# name="TenantAccountView",
# application_code="app000019",
# site_url="/tenant/accounting",
# application_type="Dash",
# description="Individual Page for tenant account view",
# ),
# ]
# for list_of_created_app in list_of_created_apps:
# created_page = Applications.find_or_create(
# **list_of_created_app,
# db=db_session,
# is_confirmed=True,
# )
# if created_page.meta_data.created:
# created_page.save(db=db_session)
# application_occupant_created = Application2Occupant.find_or_create(
# build_living_space_id=super_user.id,
# build_living_space_uu_id=str(super_user.uu_id),
# site_url=created_page.site_url,
# application_code=created_page.application_code,
# application_id=created_page.id,
# application_uu_id=str(created_page.uu_id),
# is_confirmed=True,
# db=db_session,
# )
# if application_occupant_created.meta_data.created:
# application_occupant_created.save(db=db_session)

42
ApiServices/DealerService/init_service_to_events.py
View File

@@ -11,11 +11,7 @@ from Schemas import (
)
def init_service_to_event_matches_for_super_user(super_user, db_session=None) -> None:
service_match = Services.filter_one(
Services.service_name == "Super User",
db=db_session,
).data
def init_service_to_event_matches_for_super_user(service_match: Services, employee_match: Employees, db_session=None) -> None:
list_of_all_events = Events.filter_all(db=db_session).data
Service2Events.filter_all(db=db_session).query.delete()
Service2Events.save(db=db_session)
@@ -29,10 +25,9 @@ def init_service_to_event_matches_for_super_user(super_user, db_session=None) ->
if service_to_event_found.data:
service_to_event_found.destroy(db=db_session)
print(
f"UUID: {service_to_event_found.uu_id} event is deleted from {service_match.uu_id}"
f"UUID: {service_to_event_found.uu_id} event is deleted from {service_match.service_description}"
)
created_service = Service2Events.find_or_create(
added_service = Service2Events.find_or_create(
service_id=service_match.id,
service_uu_id=str(service_match.uu_id),
event_id=list_of_event_code.id,
@@ -41,23 +36,22 @@ def init_service_to_event_matches_for_super_user(super_user, db_session=None) ->
active=True,
db=db_session,
)
if created_service.meta_data.created:
created_service.save(db=db_session)
if added_service.meta_data.created:
added_service.save(db=db_session)
print(
f"UUID: {created_service.uu_id} event is saved to {service_match.uu_id}"
f"UUID: {added_service.uu_id} event is saved to {service_match.service_description}"
)
employee_added_service = Event2Employee.find_or_create(
event_service_id=service_match.id,
event_service_uu_id=str(service_match.uu_id),
employee_id=super_user.id,
employee_uu_id=str(super_user.uu_id),
is_confirmed=True,
db=db_session,
employee_added_service = Event2Employee.find_or_create(
event_service_id=service_match.id,
event_service_uu_id=str(service_match.uu_id),
employee_id=employee_match.id,
employee_uu_id=str(employee_match.uu_id),
is_confirmed=True,
db=db_session,
)
if employee_added_service.meta_data.created:
employee_added_service.save(db=db_session)
print(
f"UUID: {employee_added_service.uu_id} event is saved to employee {employee_match.uu_id}"
)
if employee_added_service.meta_data.created:
employee_added_service.save(db=db_session)
print(
f"UUID: {employee_added_service.uu_id} event is saved to {super_user.uu_id}"
)

2
ApiServices/IdentityService/Endpoints/user/route.py
View File

@@ -69,4 +69,4 @@ def user_update_route(
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
FoundCluster = UserRouterCluster.get_event_cluster("UserUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
return event_cluster_matched.event_callable(data=data)

174
ApiServices/ManagementService/Endpoints/application/route.py
View File

@@ -7,45 +7,113 @@ from ApiControllers.providers.token_provider import TokenProvider
from Controllers.Postgres.pagination import PaginateOnly, Pagination, PaginationResult
from Controllers.Postgres.response import EndpointResponse, CreateEndpointResponse
from Schemas import Applications
from Validations.application.validations import (
RequestApplication,
)
from Validations.application.validations import RequestApplication
from Events.application.cluster import ApplicationRouterCluster
from Validations.application.validations import AddRemoveService
# Create API router
application_route = APIRouter(prefix="/application", tags=["Application Management"])
@application_route.post(
path="/list",
description="List applications endpoint",
operation_id="3189a049-bdb0-49f3-83ff-feb8cb4cb57a",
path="/list/all",
description="List all applications endpoint",
operation_id="fe30481d-802c-4490-897f-a4e95310e6bc",
)
def application_list_route(
list_options: PaginateOnly,
def application_list_all_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List applications with pagination and filtering options
List all applications with pagination and filtering options
"""
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
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,
)
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
FoundCluster = ApplicationRouterCluster.get_event_cluster("ApplicationListAll")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(list_options=data)
@application_route.post(
path="/list/available",
description="List available applications endpoint",
operation_id="7492bb02-a074-4320-b58c-4bc7d9fba3a6",
)
def application_list_available_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List available applications with pagination and filtering options
"""
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 = ApplicationRouterCluster.get_event_cluster("ApplicationListAvailable")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(list_options=data)
@application_route.post(
path="/list/appended",
description="List appended applications endpoint",
operation_id="ea7bbd58-da09-407c-a630-c324e0272385",
)
def application_list_appended_route(
data: PaginateOnly,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
List appended applications with pagination and filtering options
"""
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 = ApplicationRouterCluster.get_event_cluster("ApplicationListAppended")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(list_options=data)
@application_route.post(
path="/register/service",
description="Register event to service endpoint",
operation_id="92e0870f-f8bb-4879-ba03-c2901cc70ecc",
)
def application_register_service_route(
data: AddRemoveService,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Register event to service
"""
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 = ApplicationRouterCluster.get_event_cluster("ApplicationRegisterService")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@application_route.post(
path="/unregister/service",
description="Unregister event from service endpoint",
operation_id="7fa1a183-4c99-4746-b675-148f65ad7ba7",
)
def application_unregister_service_route(
data: AddRemoveService,
headers: CommonHeaders = Depends(CommonHeaders.as_dependency),
):
"""
Unregister event from service
"""
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 = ApplicationRouterCluster.get_event_cluster("ApplicationUnRegisterService")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@application_route.post(
@@ -61,26 +129,11 @@ def application_create_route(
Create a new application
"""
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
with Applications.new_session() as db_session:
created_application_dict = data.model_dump()
created_application = Applications.find_or_create(
db=db_session,
include_args=[
Applications.application_for,
Applications.application_code,
Applications.site_url,
],
**created_application_dict,
)
if created_application.meta_data.created:
return CreateEndpointResponse(
message="MSG0001-INSERT",
data=created_application,
).response
return CreateEndpointResponse(
message="MSG0002-FOUND",
data=created_application,
).response
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
FoundCluster = ApplicationRouterCluster.get_event_cluster("ApplicationCreate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@application_route.post(
@@ -97,27 +150,11 @@ def application_update_route(
Update an existing application
"""
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
with Applications.new_session() as db_session:
updated_application_dict = data.model_dump(exclude_unset=True, exclude_none=True)
found_application = Applications.filter_one(
Applications.uu_id == application_uuid, db=db_session
).data
if not found_application:
return CreateEndpointResponse(
message="MSG0002-FOUND",
data=found_application,
).response
updated_application = found_application.update(db=db_session,**updated_application_dict)
updated_application.save(db_session)
if updated_application.meta_data.updated:
return CreateEndpointResponse(
message="MSG0003-UPDATE",
data=updated_application,
).response
return CreateEndpointResponse(
message="MSG0003-UPDATE",
data=updated_application,
).response
event_founder_dict = dict(endpoint_code=headers.operation_id, token=token_object)
event_key = TokenProvider.retrieve_event_codes(**event_founder_dict)
FoundCluster = ApplicationRouterCluster.get_event_cluster("ApplicationUpdate")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data, uu_id=application_uuid)
@application_route.post(
@@ -158,4 +195,3 @@ def application_bind_occupant_route(
FoundCluster = ApplicationRouterCluster.get_event_cluster("ApplicationBindOccupant")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

20
ApiServices/ManagementService/Endpoints/event_endpoints/route.py
View File

@@ -5,7 +5,11 @@ from ApiControllers.abstracts.default_validations import CommonHeaders
from ApiControllers.providers.token_provider import TokenProvider
from Controllers.Postgres.pagination import PaginateOnly, Pagination, PaginationResult
from Controllers.Postgres.response import EndpointResponse
from Validations.service_endpoints.validations import Event2Employee, Event2Occupant, AddRemoveService
from Validations.service_endpoints.validations import (
Event2Employee,
Event2Occupant,
AddRemoveService,
)
from Events.event_endpoints.cluster import EventsEndpointRouterCluster
# Create API router
@@ -50,7 +54,7 @@ def event_list_appended_route(
FoundCluster = EventsEndpointRouterCluster.get_event_cluster("EventsListAppended")
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(list_options=data)
@event_endpoint_route.post(
path="/register/service",
@@ -87,7 +91,9 @@ def event_unregister_service_route(
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("EventUnregisterService")
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(
"EventUnregisterService"
)
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@@ -107,7 +113,9 @@ def event_bind_employee_extra_route(
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("EventBindEmployeeExtra")
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(
"EventBindEmployeeExtra"
)
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)
@@ -127,6 +135,8 @@ def event_bind_occupant_extra_route(
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("EventBindOccupantExtra")
FoundCluster = EventsEndpointRouterCluster.get_event_cluster(
"EventBindOccupantExtra"
)
event_cluster_matched = FoundCluster.match_event(event_key=event_key)
return event_cluster_matched.event_callable(data=data)

7
ApiServices/ManagementService/Endpoints/routes.py
View File

@@ -7,7 +7,12 @@ def get_routes() -> list[APIRouter]:
from .service_managements.route import service_management_route
from .event_endpoints.route import event_endpoint_route
return [application_route, service_endpoint_route, service_management_route, event_endpoint_route]
return [
application_route,
service_endpoint_route,
service_management_route,
event_endpoint_route,
]
def get_safe_endpoint_urls() -> list[tuple[str, str]]:

6
ApiServices/ManagementService/Events/__init__.py
View File

@@ -2,4 +2,8 @@ from .service_endpoints.cluster import ServiceEndpointRouterCluster
from .event_endpoints.cluster import EventsEndpointRouterCluster
from .application.cluster import ApplicationRouterCluster
__all__ = ["ServiceEndpointRouterCluster", "EventsEndpointRouterCluster", "ApplicationRouterCluster"]
__all__ = [
"ServiceEndpointRouterCluster",
"EventsEndpointRouterCluster",
"ApplicationRouterCluster",
]

52
ApiServices/ManagementService/Events/application/cluster.py
View File

@@ -1,18 +1,40 @@
from ApiControllers.abstracts.event_clusters import EventCluster, RouterCluster
from .supers_events import (
ApplicationListEvent,
ApplicationListAllEvent,
ApplicationListAvailableEvent,
ApplicationListAppendedEvent,
ApplicationRegisterServiceEvent,
ApplicationUnRegisterServiceEvent,
ApplicationCreateEvent,
ApplicationUpdateEvent,
ApplicationBindEmployeeEvent,
ApplicationBindOccupantEvent,
)
ApplicationRouterCluster = RouterCluster(name="ApplicationRouterCluster")
ApplicationEventClusterList = EventCluster(
name="ApplicationList", endpoint_uu_id="3189a049-bdb0-49f3-83ff-feb8cb4cb57a"
ApplicationEventClusterListAll = EventCluster(
name="ApplicationListAll", endpoint_uu_id="fe30481d-802c-4490-897f-a4e95310e6bc"
)
ApplicationEventClusterList.add_event(ApplicationListEvent)
ApplicationEventClusterListAll.add_event(ApplicationListAllEvent)
ApplicationEventClusterListAvailable = EventCluster(
name="ApplicationListAvailable", endpoint_uu_id="7492bb02-a074-4320-b58c-4bc7d9fba3a6"
)
ApplicationEventClusterListAvailable.add_event(ApplicationListAvailableEvent)
ApplicationEventClusterListAppended = EventCluster(
name="ApplicationListAppended", endpoint_uu_id="ea7bbd58-da09-407c-a630-c324e0272385"
)
ApplicationEventClusterListAppended.add_event(ApplicationListAppendedEvent)
ApplicationEventClusterRegisterService = EventCluster(
name="ApplicationRegisterService", endpoint_uu_id="92e0870f-f8bb-4879-ba03-c2901cc70ecc"
)
ApplicationEventClusterRegisterService.add_event(ApplicationRegisterServiceEvent)
ApplicationEventClusterUnregisterService = EventCluster(
name="ApplicationUnRegisterService", endpoint_uu_id="7fa1a183-4c99-4746-b675-148f65ad7ba7"
)
ApplicationEventClusterUnregisterService.add_event(ApplicationUnRegisterServiceEvent)
ApplicationEventClusterCreate = EventCluster(
name="ApplicationCreate", endpoint_uu_id="5570be78-030a-438e-8674-7e751447608b"
@@ -24,18 +46,10 @@ ApplicationEventClusterUpdate = EventCluster(
)
ApplicationEventClusterUpdate.add_event(ApplicationUpdateEvent)
ApplicationEventClusterBindEmployee = EventCluster(
name="ApplicationBindEmployee", endpoint_uu_id="2bab94fa-becb-4d8e-80f1-f4631119a521"
)
ApplicationEventClusterBindEmployee.add_event(ApplicationBindEmployeeEvent)
ApplicationEventClusterBindOccupant = EventCluster(
name="ApplicationBindOccupant", endpoint_uu_id="fccf1a59-0650-4e5c-ba8d-f389dadce01c"
)
ApplicationEventClusterBindOccupant.add_event(ApplicationBindOccupantEvent)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterList)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterListAvailable)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterListAppended)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterListAll)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterRegisterService)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterUnregisterService)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterCreate)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterUpdate)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterBindEmployee)
ApplicationRouterCluster.set_event_cluster(ApplicationEventClusterBindOccupant)

324
ApiServices/ManagementService/Events/application/supers_events.py
View File

@@ -1,3 +1,5 @@
from typing import Any
from ApiControllers.abstracts.event_clusters import Event
from Controllers.Postgres.pagination import Pagination, PaginationResult, PaginateOnly
from Controllers.Postgres.response import EndpointResponse
@@ -5,16 +7,53 @@ from Schemas import (
Applications,
Application2Employee,
Application2Occupant,
Service2Application,
Services,
)
# List endpoint
ApplicationListEvent = Event(
name="application_list",
key="b4efda1e-bde7-4659-ab1a-ef74c0fd88b6",
# 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 applications endpoint",
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
@@ -35,114 +74,247 @@ ApplicationUpdateEvent = Event(
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
"""
def application_list_all_callable(list_options: PaginateOnly):
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
)
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="",
)
return EndpointResponse(
message="MSG0003-LIST",
pagination_result=pagination_result,
).response
pagination_result = PaginationResult(data=applications_list, pagination=pagination)
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
ApplicationListEvent.event_callable = application_list_callable
ApplicationListAllEvent.event_callable = application_list_all_callable
def application_create_callable():
def application_list_available_callable(list_options: PaginateOnly):
"""
Example callable method
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
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:
service2applications = Service2Application.filter_all(
*Service2Application.convert({"service_uu_id__ilike": service_uu_id}),
db=db_session,
)
already_events = [
service_to_application.application_id for service_to_application in service2applications.data
]
if list_options.query:
applications_list = Applications.filter_all(
*Applications.convert(list_options.query), Applications.id.not_in(already_events), db=db_session
)
else:
applications_list = Applications.filter_all(Applications.id.not_in(already_events), db=db_session)
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 Applications.new_session() as db_session:
service2applications = Service2Application.filter_all(
*Service2Application.convert({"service_uu_id__ilike": service_uu_id}),
db=db_session,
)
already_events = [
service_to_application.application_id for service_to_application in service2applications.data
]
if list_options.query:
applications_list = Applications.filter_all(
*Applications.convert(list_options.query), Applications.id.in_(already_events), db=db_session
)
else:
applications_list = Applications.filter_all(Applications.id.in_(already_events), db=db_session)
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()
created_application = Applications.find_or_create(
db=db_session,
include_args=[Applications.application_for, Applications.application_code, Applications.site_url],
**created_application_dict,
)
if created_application.meta_data.created:
created_application.save(db=db_session)
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():
def application_update_callable(data: Any, uu_id: str):
"""
Example callable method
Update an existing application
"""
return {
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
with Applications.new_session() as db_session:
updated_application_dict = data.model_dump(
exclude_unset=True, exclude_none=True
)
found_application = Applications.filter_one(
Applications.uu_id == uu_id, db=db_session
).data
if not found_application:
return {
"completed": False,
"message": "MSG0002-FOUND",
"data": found_application,
}
updated_application = found_application.update(
db=db_session, **updated_application_dict
)
updated_application.save(db_session)
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_bind_employee_callable():
def application_register_service_callable(data: Any):
"""
Example callable method
Register an application to a service
"""
with Applications.new_session() as db_session:
event = Applications.filter_one_system(Applications.uu_id == data.application_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_application = Service2Application.find_or_create(
db=db_session,
include_args=[Service2Application.service_uu_id, Service2Application.application_uu_id],
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,
)
if not service_to_application.meta_data.created:
return {
"message": "MSG0003-ALREADY-FOUND",
"data": data.model_dump(),
"completed": False,
}
service_to_application.save(db=db_session)
return {
"message": "MSG0003-REGISTER",
"data": data.model_dump(),
"completed": True,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationBindEmployeeEvent.event_callable = application_bind_employee_callable
ApplicationRegisterServiceEvent.event_callable = application_register_service_callable
def application_bind_occupant_callable():
def application_unregister_service_callable(data: Any):
"""
Example callable method
Unregister an application from a service
"""
with Applications.new_session() as db_session:
application = Applications.filter_one_system(Applications.uu_id == data.application_uu_id, db=db_session)
if not application.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_application = Service2Application.filter_one_system(
Service2Application.service_id == service.data.id,
Service2Application.application_id == application.data.id,
db=db_session,
)
if not service_to_application.data:
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,
"message": "Example callable method 2",
"info": {
"host": "example_host",
"user_agent": "example_user_agent",
},
}
ApplicationBindOccupantEvent.event_callable = application_bind_occupant_callable
ApplicationUnRegisterServiceEvent.event_callable = application_unregister_service_callable

12
ApiServices/ManagementService/Events/event_endpoints/cluster.py
View File

@@ -43,6 +43,12 @@ EventsEndpointEventClusterBindOccupantExtra.add_event(EventBindOccupantExtraEven
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterListAvailable)
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterListAppended)
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterRegisterService)
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterUnregisterService)
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterBindEmployeeExtra)
EventsEndpointRouterCluster.set_event_cluster(EventsEndpointEventClusterBindOccupantExtra)
EventsEndpointRouterCluster.set_event_cluster(
EventsEndpointEventClusterUnregisterService
)
EventsEndpointRouterCluster.set_event_cluster(
EventsEndpointEventClusterBindEmployeeExtra
)
EventsEndpointRouterCluster.set_event_cluster(
EventsEndpointEventClusterBindOccupantExtra
)

69
ApiServices/ManagementService/Events/event_endpoints/supers_events.py
View File

@@ -72,22 +72,33 @@ 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)
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(list_options.query), db=db_session)
already_events = [service_to_event.event_id for service_to_event in service2events.data]
list_options.query.pop('service_uu_id__ilike', None)
list_options.query.pop('service_uu_id', None)
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)
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)
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)
@@ -105,22 +116,34 @@ 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)
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(list_options.query), db=db_session)
already_events = [service_to_event.event_id for service_to_event in service2events.data]
list_options.query.pop('service_uu_id__ilike', None)
list_options.query.pop('service_uu_id', None)
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)
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)
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)
@@ -139,10 +162,9 @@ def event_register_service_callable(data: Any):
"""
with Events.new_session() as db_session:
event = Events.filter_one_system(
Events.uu_id == data.event_uu_id,
db=db_session
Events.uu_id == data.event_uu_id, db=db_session
)
print('event', event.data)
print("event", event.data)
if not event.data:
return {
"message": "MSG0003-NOT-FOUND",
@@ -150,10 +172,9 @@ def event_register_service_callable(data: Any):
"completed": False,
}
service = Services.filter_one_system(
Services.uu_id == data.service_uu_id,
db=db_session
Services.uu_id == data.service_uu_id, db=db_session
)
print('service', service.data)
print("service", service.data)
if not service.data:
return {
"message": "MSG0003-NOT-FOUND",
@@ -169,10 +190,10 @@ def event_register_service_callable(data: Any):
service_id=service.data.id,
event_id=event.data.id,
is_confirmed=True,
service_uu_id=str(service.data.uu_id),
service_uu_id=str(service.data.uu_id),
event_uu_id=str(event.data.uu_id),
)
print('service_to_event', service_to_event)
print("service_to_event", service_to_event)
if not service_to_event.meta_data.created:
return {
"message": "MSG0003-ALREADY-FOUND",
@@ -214,8 +235,8 @@ def event_unregister_service_callable(data: Any):
"completed": False,
}
service_to_event = Service2Events.filter_one_system(
Service2Events.service_id==service.data.id,
Service2Events.event_id==event.data.id,
Service2Events.service_id == service.data.id,
Service2Events.event_id == event.data.id,
db=db_session,
)
if not service_to_event.data:

16
ApiServices/ManagementService/Events/service_endpoints/supers_events.py
View File

@@ -21,13 +21,17 @@ def service_endpoint_list_callable(data: PaginateOnly):
list_options = PaginateOnly(**data.model_dump())
with Services.new_session() as db_session:
if data.query:
services_list = Services.filter_all_system(*Services.convert(data.query), db=db_session)
services_list = Services.filter_all_system(
*Services.convert(data.query), db=db_session
)
else:
services_list = Services.filter_all_system(db=db_session)
pagination = Pagination(data=services_list)
pagination.change(**data.model_dump())
pagination_result = PaginationResult(data=services_list, pagination=pagination)
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
return EndpointResponse(
message="MSG0003-LIST", pagination_result=pagination_result
).response
ServiceEndpointListEvent.event_callable = service_endpoint_list_callable
@@ -49,13 +53,17 @@ def service_endpoint_to_events_callable(data: PaginateOnly):
list_options = PaginateOnly(**data.model_dump())
with Service2Events.new_session() as db_session:
if data.query:
services_list = Service2Events.filter_all_system(*Service2Events.convert(data.query), db=db_session)
services_list = Service2Events.filter_all_system(
*Service2Events.convert(data.query), db=db_session
)
else:
services_list = Service2Events.filter_all_system(db=db_session)
pagination = Pagination(data=services_list)
pagination.change(**data.model_dump())
pagination_result = PaginationResult(data=services_list, pagination=pagination)
return EndpointResponse(message="MSG0003-LIST", pagination_result=pagination_result).response
return EndpointResponse(
message="MSG0003-LIST", pagination_result=pagination_result
).response
ServiceEndpointToEventsEvent.event_callable = service_endpoint_to_events_callable

7
ApiServices/ManagementService/Validations/application/validations.py
View File

@@ -13,3 +13,10 @@ class RequestApplication(BaseModel):
)
application_for: str = Field(..., description="Application for (EMP, OCC)")
description: Optional[str] = Field(None, description="Application description")
class AddRemoveService(BaseModel):
"""Base model for add/remove service data"""
application_uu_id: str = Field(..., description="Application UUID")
service_uu_id: str = Field(..., description="Service UUID")

1
ApiServices/ManagementService/Validations/service_endpoints/validations.py
View File

@@ -1,5 +1,6 @@
from pydantic import BaseModel
class Event2Employee(BaseModel):
pass