from typing import Union from api_events.events.events.events_bind_modules import ModulesBindOccupantEventMethods from databases import ( Modules, BuildParts, Build, BuildLivingSpace, OccupantTypes, People, ) from api_events.events.abstract_class import MethodToEvent, ActionsSchema from api_objects.auth.token_objects import EmployeeTokenObject, OccupantTokenObject from api_validations.core_response import AlchemyJsonResponse from api_validations.validations_request import ( InsertBuildLivingSpace, UpdateBuildLivingSpace, ListOptions, ) class BuildingLivingSpacesPartsListEventMethods(MethodToEvent): event_type = "SELECT" __event_keys__ = { "36961d8a-cefa-46cc-9f7c-9d841d6351b6": "building_live_space_list", } @classmethod def building_build_parts_list( cls, list_options: ListOptions, token_dict: Union[EmployeeTokenObject, OccupantTokenObject], ): records, Build.filter_attr = [], None if isinstance(token_dict, OccupantTokenObject): occupants_build_id = Build.filter_one( Build.id == token_dict.selected_occupant.build_id, *Build.valid_record_args(Build), ).data if not occupants_build_id: Build.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"Occupant has no build registered in the system. Contact with your company.", data={}, ) occupants_build_parts = BuildParts.filter_all( BuildParts.build_id.in_(occupants_build_id.id), *BuildParts.valid_record_args(BuildParts), ).data if not occupants_build_parts: Build.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"Occupant has no build parts registered in the system. Contact with your company.", data={}, ) BuildLivingSpace.pre_query = BuildLivingSpace.filter_all( BuildLivingSpace.build_parts_id.in_( [build_part.id for build_part in occupants_build_parts] ), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ).query BuildLivingSpace.filter_attr = list_options records = BuildLivingSpace.filter_all( *BuildLivingSpace.get_smart_query(smart_query=list_options.query), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ) elif isinstance(token_dict, EmployeeTokenObject): build_id_list_query = Build.select_action( employee_id=token_dict.selected_company.employee_id ) if not build_id_list_query: Build.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"Employee has no build registered in the system. Contact with your supervisor.", data={}, ) build_part_id_list_query = BuildParts.filter_all( BuildParts.build_id.in_( [build.id for build in build_id_list_query.all()] ), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ).data if not build_part_id_list_query: Build.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"Employee has no build parts registered in the system. Contact with your supervisor.", data={}, ) BuildLivingSpace.pre_query = BuildLivingSpace.filter_all( BuildLivingSpace.build_parts_id.in_( [build_part.id for build_part in build_part_id_list_query] ), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ).query Build.filter_attr = list_options records = BuildLivingSpace.filter_all( *BuildLivingSpace.get_smart_query(smart_query=list_options.query), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ) return AlchemyJsonResponse( completed=True, message="Building Living Spaces are listed successfully", result=records, ) class BuildingLivingSpacesPartsCreateEventMethods(MethodToEvent): event_type = "CREATE" __event_keys__ = { "46d90119-3b23-4784-8053-fe11da4a3584": "building_live_space_create" } @classmethod def building_live_space_create( cls, data: InsertBuildLivingSpace, token_dict: Union[EmployeeTokenObject, OccupantTokenObject], ): from api_library.date_time_actions.date_functions import system_arrow data_dict = data.dump() build_id_list_query = Build.select_action( employee_id=token_dict.selected_company.employee_id ) build_part = BuildParts.filter_one( BuildParts.uu_id == data.build_parts_uu_id, BuildParts.build_id.in_([build.id for build in build_id_list_query.all()]), *BuildLivingSpace.valid_record_args(BuildLivingSpace), ).data if not build_part: BuildLivingSpace.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message=f"{data.build_parts_uu_id} - Build Part is not found in database. Check build part uu_id", data={ "life_person_uu_id": data.life_person_uu_id, }, ) life_person = People.filter_one( People.uu_id == data.person_uu_id or "", *People.valid_record_args(People), ).data if not life_person: BuildLivingSpace.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message=f"{data.life_person_uu_id} - Living Person is not found in database.", data={ "life_person_uu_id": data.life_person_uu_id, }, ) occupant_type = OccupantTypes.filter_by_one(uu_id=data.occupant_type_uu_id).data if not occupant_type: BuildLivingSpace.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message=f"{data.occupant_type_uu_id} - Occupant Type is not found in database. Check occupant type uu_id", data={ "occupant_type_uu_id": data.occupant_type_uu_id, }, ) data_dict["occupant_type"] = occupant_type.id data_dict["occupant_type_uu_id"] = str(occupant_type.uu_id) data_dict["build_parts_id"] = build_part.id data_dict["build_parts_uu_id"] = str(build_part.uu_id) data_dict["person_id"] = life_person.id data_dict["person_uu_id"] = str(life_person.uu_id) living_space_id = BuildLivingSpace.select_only( BuildLivingSpace.build_parts_id == build_part.id, BuildLivingSpace.person_id == life_person.id, BuildLivingSpace.occupant_type == occupant_type.id, *BuildLivingSpace.valid_record_args(BuildLivingSpace), select_args=[BuildLivingSpace.id], order_by=BuildLivingSpace.expiry_starts.desc(), limit=1, ).data last_living_space = BuildLivingSpace.filter_one( BuildLivingSpace.id == living_space_id[0] if living_space_id else None, *BuildLivingSpace.valid_record_args(BuildLivingSpace), ) data_dict["expiry_starts"] = str(system_arrow.now()) created_living_space = BuildLivingSpace.create_action( data=data_dict, token_dict=token_dict ) if last_living_space: if last_living_space.expiry_ends > str(system_arrow.now()): last_living_space.expiry_ends = str(system_arrow.shift(minutes=-10)) last_living_space.save() user_module = Modules.filter_one( Modules.module_code == "USR-PUB", system=True ).data ModulesBindOccupantEventMethods.modules_bind_occupant_system( build_living_space_id=created_living_space.id, modules_id=user_module.id, ) created_living_space.save() return created_living_space class BuildingLivingSpacesPartsUpdateEventMethods(MethodToEvent): event_type = "UPDATE" __event_keys__ = { "c786e15c-c03e-4e8f-936c-7e5e5ec9bbcc": "building_live_space_update", } @classmethod def building_live_space_update( cls, build_uu_id: str, data: UpdateBuildLivingSpace, token_dict: Union[EmployeeTokenObject, OccupantTokenObject], ): from api_library.date_time_actions.date_functions import system_arrow if isinstance(token_dict, OccupantTokenObject): data_dict = data.dump() build_id_list_query = Build.select_action( employee_id=token_dict.selected_company.employee_id ) build_part = BuildParts.filter_one( BuildParts.uu_id == data.build_parts_uu_id, BuildParts.build_id.in_( [build.id for build in build_id_list_query.all()] ), *BuildParts.valid_record_args(BuildParts), ).data if not build_part: BuildLivingSpace.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"{data.life_person_uu_id} - Living Person is not found in database.", data={ "person_uu_id": data.life_person_uu_id, }, ) life_person = People.filter_one( People.uu_id == data.life_person_uu_id or "" ).data if not life_person: BuildLivingSpace.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message=f"{data.life_person_uu_id} - Living Person is not found in database.", data={ "person_uu_id": data.life_person_uu_id, }, ) living_space_id = BuildLivingSpace.select_only( *BuildLivingSpace.valid_record_args(BuildLivingSpace), select_args=[BuildLivingSpace.id], order_by=BuildLivingSpace.expiry_starts.desc(), limit=1, ).get(1) last_living_space = BuildLivingSpace.filter_one( BuildLivingSpace.id == living_space_id if living_space_id else None, *BuildLivingSpace.valid_record_args(BuildLivingSpace), ).data data_dict["expiry_starts"] = str(system_arrow.now()) data_dict["is_tenant_live"] = bool(data.is_tenant_live) data_dict["build_parts_id"] = build_part.id if data_dict["is_tenant_live"]: owner_person = getattr(last_living_space, "owner_person_id", None) if not owner_person: BuildLivingSpace.raise_http_exception( status_code="HTTP_403_FORBIDDEN", error_case="UNAUTHORIZED", message="Owner person of build part is not defined. Please register owner of part first.", data=build_part.get_dict(), ) data_dict["life_person_id"] = life_person.id data_dict["owner_person_id"] = owner_person else: data_dict["life_person_id"] = life_person.id data_dict["owner_person_id"] = life_person.id del data_dict["build_parts_uu_id"], data_dict["life_person_uu_id"] BuildingLivingSpacesPartsListEventMethod = BuildingLivingSpacesPartsListEventMethods( action=ActionsSchema(endpoint="/building/living_space/list") ) BuildingLivingSpacesPartsCreateEventMethod = ( BuildingLivingSpacesPartsCreateEventMethods( action=ActionsSchema(endpoint="/building/living_space/create") ) ) BuildingLivingSpacesPartsUpdateEventMethod = ( BuildingLivingSpacesPartsUpdateEventMethods( action=ActionsSchema(endpoint="/building/living_space/update") ) )