import typing from ApiValidations.Custom.token_objects import ( OccupantTokenObject, EmployeeTokenObject, ) from Schemas import ( BuildLivingSpace, AccountRecords, BuildIbans, BuildDecisionBookPayments, ApiEnumDropdown, ) from ApiLibrary import system_arrow from ApiValidations.Request import ( InsertAccountRecord, UpdateAccountRecord, ListOptions, ) from Services.PostgresDb.Models.alchemy_response import AlchemyJsonResponse from ApiEvents.abstract_class import ( MethodToEvent, RouteFactoryConfig, EndpointFactoryConfig, ) from ApiValidations.Response import AccountRecordResponse class AccountRecordsListEventMethods(MethodToEvent): event_type = "SELECT" event_description = "" event_category = "" __event_keys__ = { "7192c2aa-5352-4e36-98b3-dafb7d036a3d": "account_records_list", "208e6273-17ef-44f0-814a-8098f816b63a": "account_records_list_flt_res", } __event_validation__ = { "7192c2aa-5352-4e36-98b3-dafb7d036a3d": AccountRecordResponse, "208e6273-17ef-44f0-814a-8098f816b63a": AccountRecordResponse, } @classmethod def account_records_list( cls, list_options: ListOptions, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): db_session = AccountRecords.new_session() if isinstance(token_dict, OccupantTokenObject): AccountRecords.pre_query = AccountRecords.filter_all( AccountRecords.company_id == token_dict.selected_occupant.responsible_company_id, db=db_session, ).query elif isinstance(token_dict, EmployeeTokenObject): AccountRecords.pre_query = AccountRecords.filter_all( AccountRecords.company_id == token_dict.selected_company.company_id, db=db_session, ).query AccountRecords.filter_attr = list_options records = AccountRecords.filter_all(db=db_session) return AlchemyJsonResponse( completed=True, message="Account records listed successfully", result=records, cls_object=AccountRecords, filter_attributes=list_options, response_model=AccountRecordResponse, ) @classmethod def account_records_list_flt_res( cls, list_options: ListOptions, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): db_session = AccountRecords.new_session() if not isinstance(token_dict, OccupantTokenObject): raise AccountRecords.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message="Only Occupant can see this data", data={}, ) return_list = [] living_space: BuildLivingSpace = BuildLivingSpace.filter_by_one( id=token_dict.selected_occupant.living_space_id ).data if not living_space: raise AccountRecords.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message="Living space not found", data={}, ) if not list_options: list_options = ListOptions() main_filters = [ AccountRecords.living_space_id == token_dict.selected_occupant.living_space_id, BuildDecisionBookPayments.process_date >= str(system_arrow.now().shift(months=-3).date()), BuildDecisionBookPayments.process_date < str(system_arrow.find_last_day_of_month(living_space.expiry_ends)), BuildDecisionBookPayments.process_date >= str(system_arrow.get(living_space.expiry_starts)), BuildDecisionBookPayments.is_confirmed == True, AccountRecords.active == True, ] order_type = "desc" if list_options.order_type: order_type = "asc" if list_options.order_type[0] == "a" else "desc" order_by_list = BuildDecisionBookPayments.process_date.desc() if list_options.order_field: if list_options.order_field == "process_date": order_by_list = ( BuildDecisionBookPayments.process_date.asc() if order_type == "asc" else BuildDecisionBookPayments.process_date.desc() ) if list_options.order_field == "bank_date": order_by_list = ( AccountRecords.bank_date.desc() if order_type == "asc" else AccountRecords.bank_date.asc() ) if list_options.order_field == "currency_value": order_by_list = ( AccountRecords.currency_value.desc() if order_type == "asc" else AccountRecords.currency_value.asc() ) if list_options.order_field == "process_comment": order_by_list = ( AccountRecords.process_comment.desc() if order_type == "asc" else AccountRecords.process_comment.asc() ) if list_options.order_field == "payment_amount": order_by_list = ( BuildDecisionBookPayments.payment_amount.desc() if order_type == "asc" else BuildDecisionBookPayments.payment_amount.asc() ) if list_options.query: for key, value in list_options.query.items(): if key == "process_date": main_filters.append(BuildDecisionBookPayments.process_date == value) if key == "bank_date": main_filters.append(AccountRecords.bank_date == value) if key == "currency": main_filters.append(BuildDecisionBookPayments.currency == value) if key == "currency_value": main_filters.append(AccountRecords.currency_value == value) if key == "process_comment": main_filters.append(AccountRecords.process_comment == value) if key == "payment_amount": main_filters.append( BuildDecisionBookPayments.payment_amount == value ) query = ( AccountRecords.session.query( BuildDecisionBookPayments.process_date, BuildDecisionBookPayments.payment_amount, BuildDecisionBookPayments.currency, AccountRecords.bank_date, AccountRecords.currency_value, AccountRecords.process_comment, BuildDecisionBookPayments.uu_id, ) .join( AccountRecords, AccountRecords.id == BuildDecisionBookPayments.account_records_id, ) .filter(*main_filters) ).order_by(order_by_list) query.limit(list_options.size or 5).offset( (list_options.page or 1 - 1) * list_options.size or 5 ) for list_of_values in query.all() or []: return_list.append( { "process_date": list_of_values[0], "payment_amount": list_of_values[1], "currency": list_of_values[2], "bank_date": list_of_values[3], "currency_value": list_of_values[4], "process_comment": list_of_values[5], } ) return AlchemyJsonResponse( completed=True, message="Account records listed successfully", result=return_list, cls_object=AccountRecords, filter_attributes=list_options, response_model=AccountRecordResponse, ) class AccountRecordsCreateEventMethods(MethodToEvent): event_type = "CREATE" event_description = "" event_category = "" __event_keys__ = { "31f4f32f-0cd4-4995-8a6a-f9f56335848a": "account_records_create", } __event_validation__ = { "31f4f32f-0cd4-4995-8a6a-f9f56335848a": InsertAccountRecord, } @classmethod def account_records_create( cls, data: InsertAccountRecord, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): data_dict = data.excluded_dump() if isinstance(token_dict, OccupantTokenObject): db_session = AccountRecords.new_session() build_iban = BuildIbans.filter_one( BuildIbans.iban == data.iban, BuildIbans.build_id == token_dict.selected_occupant.build_id, db=db_session, ).data if not build_iban: raise BuildIbans.raise_http_exception( status_code="HTTP_404_NOT_FOUND", error_case="UNAUTHORIZED", message=f"{data.iban} is not found in company related to your organization", data={ "iban": data.iban, }, ) account_record = AccountRecords.find_or_create(**data.excluded_dump()) return AlchemyJsonResponse( completed=True, message="Account record created successfully", result=account_record, ) elif isinstance(token_dict, EmployeeTokenObject): # Build.pre_query = Build.select_action( # employee_id=token_dict.selected_employee.employee_id, # ) # build_ids_list = Build.filter_all( # ) # build_iban = BuildIbans.filter_one( # BuildIbans.iban == data.iban, # BuildIbans.build_id.in_([build.id for build in build_ids_list.data]), # ).data # if not build_iban: # BuildIbans.raise_http_exception( # status_code="HTTP_404_NOT_FOUND", # error_case="UNAUTHORIZED", # message=f"{data.iban} is not found in company related to your organization", # data={ # "iban": data.iban, # }, # ) bank_date = system_arrow.get(data.bank_date) data_dict["bank_date_w"] = bank_date.weekday() data_dict["bank_date_m"] = bank_date.month data_dict["bank_date_d"] = bank_date.day data_dict["bank_date_y"] = bank_date.year if int(data.currency_value) < 0: debit_type = ApiEnumDropdown.filter_by_one( system=True, enum_class="DebitTypes", key="DT-D" ).data data_dict["receive_debit"] = debit_type.id data_dict["receive_debit_uu_id"] = str(debit_type.uu_id) else: debit_type = ApiEnumDropdown.filter_by_one( system=True, enum_class="DebitTypes", key="DT-R" ).data data_dict["receive_debit"] = debit_type.id data_dict["receive_debit_uu_id"] = str(debit_type.uu_id) account_record = AccountRecords.insert_one(data_dict).data return AlchemyJsonResponse( completed=True, message="Account record created successfully", result=account_record, ) class AccountRecordsUpdateEventMethods(MethodToEvent): event_type = "UPDATE" event_description = "" event_category = "" __event_keys__ = { "ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": "account_records_update", } __event_validation__ = { "ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": UpdateAccountRecord, } @classmethod def build_area_update( cls, build_uu_id: str, data: UpdateAccountRecord, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): if isinstance(token_dict, OccupantTokenObject): pass elif isinstance(token_dict, EmployeeTokenObject): pass AccountRecords.build_parts_id = token_dict.selected_occupant.build_part_id account_record = AccountRecords.update_one(build_uu_id, data).data return AlchemyJsonResponse( completed=True, message="Account record updated successfully", result=account_record, cls_object=AccountRecords, response_model=UpdateAccountRecord, ) class AccountRecordsPatchEventMethods(MethodToEvent): event_type = "PATCH" event_description = "" event_category = "" __event_keys__ = { "34c38937-42a2-45f1-b2ef-a23978650aee": "account_records_patch", } __event_validation__ = { "34c38937-42a2-45f1-b2ef-a23978650aee": None, } @classmethod def build_area_patch( cls, build_uu_id: str, data, token_dict: typing.Union[EmployeeTokenObject, OccupantTokenObject], ): account_record = AccountRecords.patch_one(build_uu_id, data).data return AlchemyJsonResponse( completed=True, message="Account record patched successfully", result=account_record, )