evyos-center-server/wag-managment-api-service-version-2
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

events imports are checked

Browse Source
This commit is contained in:
berkay
2024-11-08 15:05:12 +03:00
parent 643d6d8f65
commit a5b1e0b2f4
71 changed files with 2517 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
api_events/events/events/events_bind_events.py
View File

@@ -1,3 +1,5 @@
from typing import Union
from fastapi import status
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
@@ -8,9 +10,9 @@ from databases import (
Build,
Events,
Event2Occupant,
OccupantTypes,
OccupantTypes,
)
from api_validations import RegisterEvents2Occupant
from api_validations.validations_request import RegisterEvents2Occupant
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_validations.core_response import return_json_response_from_alchemy
@@ -24,7 +26,7 @@ class EventBindOccupantEventMethods(MethodToEvent):
@classmethod
def bind_events_occupant_super_user(
cls, data: RegisterEvents2Occupant, token_dict: EmployeeTokenObject
cls, data: RegisterEvents2Occupant, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
):
if not str(token_dict.user_type) == "1":
@@ -108,6 +110,7 @@ class EventBindOccupantEventMethods(MethodToEvent):
events_added.append(event)
if events_added:
Events.save()
return JSONResponse(
content={
"completed": True,
@@ -134,7 +137,9 @@ class EventBindEmployeeEventMethods(MethodToEvent):
}
@classmethod
def bind_events_employee(cls, data: RegisterEvents2Occupant, token_dict):
def bind_events_employee(
cls, data: RegisterEvents2Occupant, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]
):
return token_dict.available_event(data=data, token_dict=token_dict)

74
api_events/events/events/events_bind_modules.py Normal file
View File

@@ -0,0 +1,74 @@
import typing
from databases import (
Modules,
BuildLivingSpace,
)
from api_validations.validations_request import RegisterModules2Occupant, RegisterModules2Employee
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_events.events.events.events_bind_services import ServiceBindOccupantEventMethods
from api_library.date_time_actions.date_functions import system_arrow
from api_validations.core_response import return_json_response_from_alchemy
class ModulesBindOccupantEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"": "modules_bind_occupant",
}
@classmethod
def modules_bind_occupant_system(
cls, build_living_space_id: int, modules_id: int, expires_at: str = None
):
living_space = BuildLivingSpace.filter_one(Modules.id==build_living_space_id).data
modules = Modules.filter_one(Modules.id==modules_id).data
service_build_dict = dict(build_living_space_id=living_space.id)
service_build_dict["expires_at"] = str(system_arrow.get(living_space.expiry_ends))
if not expires_at:
service_build_dict["expires_at"] = str(system_arrow.get(expires_at))
for service in modules.retrieve_services():
ServiceBindOccupantEventMethods.bind_services_occupant_system(
**service_build_dict,
service_id=service.id,
)
BuildLivingSpace.save()
return True
@classmethod
def modules_bind_occupant(
cls,
data: RegisterModules2Occupant,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return
class ModulesBindEmployeeEventMethods(MethodToEvent):
event_type = "UPDATE"
__event_keys__ = {
"": "modules_bind_employee",
}
@classmethod
def modules_bind_employee(
cls,
data: RegisterModules2Employee,
token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject],
):
return
ModulesBindOccupantEventMethod = ModulesBindOccupantEventMethods(
action=ActionsSchema(endpoint="/bind/modules/occupant")
)
ModulesBindEmployeeEventMethod = ModulesBindEmployeeEventMethods(
action=ActionsSchema(endpoint="/bind/modules/employee")
)

32
api_events/events/events/events_bind_services.py
View File

@@ -15,7 +15,7 @@ from databases import (
Event2Employee,
Event2Occupant,
)
from api_validations import RegisterServices2Occupant, RegisterServices2Employee
from api_validations.validations_request import RegisterServices2Occupant, RegisterServices2Employee
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_validations.core_response import return_json_response_from_alchemy
@@ -33,21 +33,9 @@ class ServiceBindOccupantEventMethods(MethodToEvent):
):
from sqlalchemy.dialects.postgresql import insert
living_space = BuildLivingSpace.find_or_abort(
id=build_living_space_id,
)
service = Services.find_or_abort(
id=service_id,
)
default_module = Modules.find_one(module_code="USR-PUB")
add_default_service = Services.find_one(module_id=default_module.id)
service_events = Service2Events.filter_all(
Service2Events.service_id == service.id,
)
default_service_events = Service2Events.filter_all(
Service2Events.service_id == add_default_service.id,
)
add_events_list = service_events.data + default_service_events.data
living_space = BuildLivingSpace.filter_one(BuildLivingSpace.id==build_living_space_id)
service = Services.filter_one(Services.id==service_id)
add_events_list = Service2Events.filter_all(Service2Events.service_id == service.id).data
if not add_events_list:
raise Exception(
"Service has no events registered. Please contact with your manager"
@@ -74,8 +62,7 @@ class ServiceBindOccupantEventMethods(MethodToEvent):
)
count_row = session_execute.rowcount
print(f"{count_row} events are added to occupant {str(living_space.uu_id)}")
Services.session.commit()
Services.session.flush()
Services.save()
@classmethod
def bind_services_occupant(
@@ -178,8 +165,7 @@ class ServiceBindOccupantEventMethods(MethodToEvent):
)
count_row = session_execute.rowcount
print(f"{count_row} events are added to employee {str(living_space.uu_id)}")
Services.session.commit()
Services.session.flush()
Services.save()
class ServiceBindEmployeeEventMethods(MethodToEvent):
@@ -233,8 +219,7 @@ class ServiceBindEmployeeEventMethods(MethodToEvent):
)
count_row = session_execute.rowcount
print(f"{count_row} events are added to employee {employee.uu_id}")
Services.session.commit()
Services.session.flush()
Services.save()
@classmethod
def bind_services_employee_super_user(
@@ -308,9 +293,8 @@ class ServiceBindEmployeeEventMethods(MethodToEvent):
)
)
count_row = session_execute.rowcount
Services.session.commit()
Services.session.flush()
if not count_row:
Services.save()
return JSONResponse(
content={
"completed": False,

10
api_events/events/events/events_events.py
View File

@@ -1,3 +1,5 @@
from typing import Union
from fastapi.exceptions import HTTPException
from databases import (
@@ -9,7 +11,7 @@ from databases import (
Event2Employee,
BuildLivingSpace,
)
from api_validations import (
from api_validations.validations_request import (
RegisterEvents2Employee,
RegisterEvents2Occupant,
CreateEvents,
@@ -29,7 +31,7 @@ class EventsListEventMethods(MethodToEvent):
}
@classmethod
def events_list(cls, list_options: ListOptions, token_dict):
def events_list(cls, list_options: ListOptions, token_dict: Union[EmployeeTokenObject, OccupantTokenObject]):
records = Events.filter_active(
*Events.get_smart_query(list_options.query),
)
@@ -56,6 +58,7 @@ class EventsCreateEventMethods(MethodToEvent):
active=True,
deleted=False,
)
Events.save()
return {
"status": "success",
"message": "Event created successfully.",
@@ -85,6 +88,7 @@ class EventsUpdateEventMethods(MethodToEvent):
event_date=data.event_date,
event_location=data.event_location,
)
Events.save()
return {
"status": "success",
"message": "Event updated successfully.",
@@ -155,6 +159,7 @@ class EventsBindEventToOccupantMethods(MethodToEvent):
employee = Event2Employee.find_or_create(
**token_dict.user_creds, employee_id=employee.id, event_id=event.id
)
Events.save()
return {
"status": "success",
"message": "Events registered successfully.",
@@ -191,6 +196,7 @@ class EventsBindEventToEmployeeMethods(MethodToEvent):
build_living_space_id=occupant.id,
event_id=event.id,
)
Events.save()
return {
"status": "success",
"message": "Events registered successfully.",

85
api_events/events/events/events_model_entities.py
View File

@@ -1,85 +0,0 @@
from fastapi import status
from fastapi.responses import JSONResponse
from api_validations import DepartmentsPydantic, ListOptions
from databases import Events
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject
from api_validations.core_response import return_json_response_from_alchemy
class ModelEntitiesEvents(MethodToEvent):
@classmethod
def model_entities_list(cls, list_options: ListOptions, token_dict):
Events.filter_attr = list_options
records = Events.filter_active(
*Events.get_smart_query(smart_query=list_options.query),
Events.company_id == token_dict.selected_company.company_id
)
return return_json_response_from_alchemy(
response=records, pagination=list_options
)
@classmethod
def model_entities_create(cls, data: DepartmentsPydantic, token_dict):
created_events = Events.create_action(data=data, token=token_dict)
if created_events.is_found:
return JSONResponse(
content={
"completed": False,
"message": "Events record already exits here is the record",
"data": created_events.get_dict(),
},
status_code=status.HTTP_202_ACCEPTED,
)
return JSONResponse(
content={
"completed": True,
"message": "Create Events record",
"data": created_events.get_dict(),
},
status_code=status.HTTP_200_OK,
)
@classmethod
def model_entities_update(
cls, company_uu_id: str, data: DepartmentsPydantic, token_dict
):
find_one_events = Events.find_one_or_abort(uu_id=company_uu_id)
access_authorized_events = Events.select_action(
duty_id=getattr(token_dict, "duty_id", 5),
filter_expr=[Events.id == token_dict.get("")],
action="update",
data=data,
)
return JSONResponse(
content={
"completed": True,
"message": "Update Events record",
"data": access_authorized_events.get_dict(),
},
status_code=status.HTTP_200_OK,
)
@classmethod
def model_entities_patch(
cls, company_uu_id: str, data: DepartmentsPydantic, token_dict
):
find_one_events = Events.find_one_or_abort(uu_id=company_uu_id)
access_authorized_events = Events.select_action(
duty_id=getattr(token_dict, "duty_id", 5),
filter_expr=[Events.id == token_dict.get("")],
action="patch",
data=data,
)
return JSONResponse(
content={
"completed": True,
"message": "Patch Events record",
"data": access_authorized_events.get_dict(),
},
status_code=status.HTTP_200_OK,
)

2
api_events/events/events/events_models.py
View File

@@ -1,4 +1,4 @@
from validations import DepartmentsPydantic, PatchRecord, ListOptions
from api_validations.validations_request import DepartmentsPydantic, PatchRecord, ListOptions
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject

2
api_events/events/events/events_modules.py
View File

@@ -1,4 +1,4 @@
from validations import DepartmentsPydantic, PatchRecord
from api_validations.validations_request import DepartmentsPydantic, PatchRecord
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject

2
api_events/events/events/events_services.py
View File

@@ -1,4 +1,4 @@
from validations import DepartmentsPydantic, PatchRecord, ListOptions
from api_validations.validations_request import DepartmentsPydantic, PatchRecord, ListOptions
from api_events.events.abstract_class import MethodToEvent, ActionsSchema
from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject