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

more services added

Browse Source
This commit is contained in:
Berkay
2025-05-16 14:55:50 +03:00
parent 4606f0721d
commit 5f7cb35ccc
95 changed files with 2600 additions and 717 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
api_services/api_builds/management_service/events/application/supers_events.py
View File

@@ -103,10 +103,11 @@ ApplicationBindOccupantEvent = Event(
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.filter_all(*Applications.convert(list_options.query), db=db_session)
applications_list = Applications.query.filter(*Applications.convert(list_options.query))
else:
applications_list = Applications.filter_all(db=db_session)
applications_list = Applications.query.filter()
pagination = Pagination(data=applications_list)
pagination.change(**list_options.model_dump())
pagination_result = PaginationResult(data=applications_list, pagination=pagination)
@@ -131,19 +132,13 @@ def application_list_available_callable(list_options: PaginateOnly):
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
]
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.filter_all(
*Applications.convert(list_options.query), Applications.id.not_in(already_events), db=db_session
)
applications_list = Applications.query.filter(*Applications.convert(list_options.query), Applications.id.not_in(already_events))
else:
applications_list = Applications.filter_all(Applications.id.not_in(already_events), db=db_session)
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)
@@ -167,21 +162,15 @@ def application_list_appended_callable(list_options: PaginateOnly):
}
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
]
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.filter_all(
*Applications.convert(list_options.query), Applications.id.in_(already_events), db=db_session
)
applications_list = Applications.query.filter(*Applications.convert(list_options.query), Applications.id.in_(already_events))
else:
applications_list = Applications.filter_all(Applications.id.in_(already_events), db=db_session)
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)
@@ -197,13 +186,10 @@ def application_create_callable(data: Any):
"""
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)
Applications.set_session(db_session)
created_application = Applications.create(**created_application_dict)
if created_application:
created_application.save()
return {
"completed": True,
"message": "MSG0001-INSERT",
@@ -224,22 +210,16 @@ 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.filter_one(
Applications.uu_id == uu_id, db=db_session
).data
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(
db=db_session, **updated_application_dict
)
updated_application.save(db_session)
updated_application = found_application.update(**updated_application_dict)
updated_application.save()
if updated_application.meta_data.updated:
return {
"completed": True,
@@ -261,23 +241,32 @@ def application_register_service_callable(data: Any):
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:
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.filter_one_system(Services.uu_id == data.service_uu_id, db=db_session)
if not service.data:
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.find_or_create(
db=db_session,
include_args=[Service2Application.service_uu_id, Service2Application.application_uu_id],
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,
@@ -286,13 +275,7 @@ def application_register_service_callable(data: Any):
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)
service_to_application.save()
return {
"message": "MSG0003-REGISTER",
"data": data.model_dump(),
@@ -308,26 +291,26 @@ def application_unregister_service_callable(data: Any):
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:
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.filter_one_system(Services.uu_id == data.service_uu_id, db=db_session)
if not service.data:
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.filter_one_system(
service_to_application = Service2Application.query.filter(
Service2Application.service_id == service.data.id,
Service2Application.application_id == application.data.id,
db=db_session,
)
if not service_to_application.data:
).first()
if not service_to_application:
return {
"message": "MSG0003-NOT-FOUND",
"data": data.model_dump(),