""" Account records service implementation. """ from typing import Any, Union, Optional from ApiLayers.ApiLibrary import system_arrow from ApiLayers.ApiValidations.Custom.token_objects import ( OccupantTokenObject, EmployeeTokenObject, ) from ApiLayers.ApiValidations.Request import ( InsertAccountRecord, UpdateAccountRecord, ListOptions, ) from ApiLayers.Schemas import ( BuildLivingSpace, BuildDecisionBookPayments, AccountRecords, BuildIbans, ApiEnumDropdown, ) from ApiLayers.ApiValidations.Response import AccountRecordResponse from Events.base_request_model import BaseRouteModel, ListOptionsBase from Services.PostgresDb.Models.pagination import PaginationResult class AccountListEventMethods(BaseRouteModel): """ Account records list by with full privileges. Accepts List Options { "data": { "page": 1, "size": 10, "order_field": ["uu_id",] "order_type": ["desc"], "query": { "process_date__gt": "2021-09-01", } } } """ @classmethod def account_records_list( cls, data: Optional[Union[dict, ListOptions]] ) -> PaginationResult: list_options_base = ListOptionsBase( table=AccountRecords, list_options=data, model_query=None, ) db_session, query_options = list_options_base.init_list_options() if cls.context_retriever.token.is_occupant: AccountRecords.pre_query = AccountRecords.filter_all( AccountRecords.company_id == cls.context_retriever.token.selected_occupant.responsible_company_id, db=db_session, ).query elif cls.context_retriever.token.is_employee: AccountRecords.pre_query = AccountRecords.filter_all( AccountRecords.company_id == cls.context_retriever.token.selected_company.company_id, db=db_session, ).query records = AccountRecords.filter_all(*query_options.convert(), db=db_session) return list_options_base.paginated_result( records=records, response_model=getattr(cls.context_retriever, "RESPONSE_VALIDATOR", None), ) class AccountCreateEventMethods(BaseRouteModel): @classmethod def account_records_create(cls, data: Any): data_dict = data.excluded_dump() db_session = AccountRecords.new_session() if cls.context_retriever.token.is_occupant: build_iban = BuildIbans.filter_one( BuildIbans.iban == data.iban, BuildIbans.build_id == cls.context_retriever.token.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 cls.context_retriever.token.is_employee: # 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", db=db_session ).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", db=db_session ).data data_dict["receive_debit"] = debit_type.id data_dict["receive_debit_uu_id"] = str(debit_type.uu_id) account_record = AccountRecords.find_or_create( data_dict, db=db_session ).data # return AlchemyJsonResponse( # completed=True, # message="Account record created successfully", # result=account_record, # ) class AccountUpdateEventMethods(BaseRouteModel): @classmethod def account_records_update(cls, build_uu_id: str, data: Any): if cls.context_retriever.token.is_occupant: pass elif cls.context_retriever.token.is_employee: pass AccountRecords.build_parts_id = ( cls.context_retriever.token.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, # ) # @classmethod # def account_records_list_flt_res(cls, list_options: ListOptions) -> PaginationResult: # list_options_base = ListOptionsBase( # table=AccountRecords, list_options=list_options, model_query=None, # ) # db_session, query_options = list_options_base.init_list_options() # if not cls.context_retriever.token.is_occupant: # 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=cls.context_retriever.token.selected_occupant.living_space_id, db=db_session # ).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 # == cls.context_retriever.token.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, # )