From 84ad065782e11465651389ef433b8c2dd5cb6ad4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Sun, 16 Apr 2023 22:46:05 +0800 Subject: [PATCH] Merged the "accounting.utils.unapplied" module into the "accounting.utils.offset_matcher" module as the "__get_unapplied" method of the OffsetMatcher class. It is only used in the offset matcher. --- src/accounting/utils/offset_matcher.py | 50 ++++++++++++++++-- src/accounting/utils/unapplied.py | 72 -------------------------- 2 files changed, 47 insertions(+), 75 deletions(-) delete mode 100644 src/accounting/utils/unapplied.py diff --git a/src/accounting/utils/offset_matcher.py b/src/accounting/utils/offset_matcher.py index adc58cb..4d26868 100644 --- a/src/accounting/utils/offset_matcher.py +++ b/src/accounting/utils/offset_matcher.py @@ -17,11 +17,15 @@ """The forms for the unmatched offset management. """ +from decimal import Decimal + import sqlalchemy as sa from sqlalchemy.orm import selectinload +from accounting import db from accounting.models import Account, JournalEntry, JournalEntryLineItem -from accounting.utils.unapplied import get_unapplied_original_line_items +from accounting.utils.cast import be +from accounting.utils.offset_alias import offset_alias class OffsetPair: @@ -67,8 +71,7 @@ class OffsetMatcher: :return: None. """ - self.unapplied: list[JournalEntryLineItem] \ - = get_unapplied_original_line_items(self.account) + self.unapplied: list[JournalEntryLineItem] = self.__get_unapplied() self.total = len(self.unapplied) if self.total == 0: self.is_having_matches = False @@ -95,6 +98,47 @@ class OffsetMatcher: remains.remove(offset_candidates[0]) self.is_having_matches = len(self.matched_pairs) > 0 + def __get_unapplied(self) -> list[JournalEntryLineItem]: + """Returns the unapplied original line items of the account. + + :return: The unapplied original line items of the account. + """ + offset: sa.Alias = offset_alias() + net_balance: sa.Label \ + = (JournalEntryLineItem.amount + + sa.func.sum(sa.case( + (be(offset.c.is_debit == JournalEntryLineItem.is_debit), + offset.c.amount), + else_=-offset.c.amount))).label("net_balance") + select_net_balances: sa.Select \ + = sa.select(JournalEntryLineItem.id, net_balance) \ + .join(Account) \ + .join(offset, be(JournalEntryLineItem.id + == offset.c.original_line_item_id), + isouter=True) \ + .filter(be(Account.id == self.account.id), + sa.or_(sa.and_(Account.base_code.startswith("2"), + sa.not_(JournalEntryLineItem.is_debit)), + sa.and_(Account.base_code.startswith("1"), + JournalEntryLineItem.is_debit))) \ + .group_by(JournalEntryLineItem.id) \ + .having(sa.or_(sa.func.count(offset.c.id) == 0, net_balance != 0)) + net_balances: dict[int, Decimal] \ + = {x.id: x.net_balance + for x in db.session.execute(select_net_balances).all()} + line_items: list[JournalEntryLineItem] = JournalEntryLineItem.query \ + .filter(JournalEntryLineItem.id.in_({x for x in net_balances})) \ + .join(JournalEntry) \ + .order_by(JournalEntry.date, JournalEntry.no, + JournalEntryLineItem.is_debit, JournalEntryLineItem.no) \ + .options(selectinload(JournalEntryLineItem.currency), + selectinload(JournalEntryLineItem.journal_entry)).all() + for line_item in line_items: + line_item.net_balance = line_item.amount \ + if net_balances[line_item.id] is None \ + else net_balances[line_item.id] + return line_items + def __get_unmatched_offsets(self) -> list[JournalEntryLineItem]: """Returns the unmatched offsets of an account. diff --git a/src/accounting/utils/unapplied.py b/src/accounting/utils/unapplied.py deleted file mode 100644 index b6cf0b7..0000000 --- a/src/accounting/utils/unapplied.py +++ /dev/null @@ -1,72 +0,0 @@ -# The Mia! Accounting Project. -# Author: imacat@mail.imacat.idv.tw (imacat), 2023/4/8 - -# Copyright (c) 2023 imacat. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""The unapplied original line item utilities. - -""" -from decimal import Decimal - -import sqlalchemy as sa -from sqlalchemy.orm import selectinload - -from accounting import db -from accounting.models import Account, JournalEntry, JournalEntryLineItem -from accounting.utils.cast import be -from accounting.utils.offset_alias import offset_alias - - -def get_unapplied_original_line_items(account: Account) \ - -> list[JournalEntryLineItem]: - """Queries and returns the unapplied original line items in an account. - - :param account: The account. - :return: The unapplied original line items in the account. - """ - offset: sa.Alias = offset_alias() - net_balance: sa.Label \ - = (JournalEntryLineItem.amount - + sa.func.sum(sa.case( - (be(offset.c.is_debit == JournalEntryLineItem.is_debit), - offset.c.amount), - else_=-offset.c.amount))).label("net_balance") - select_net_balances: sa.Select \ - = sa.select(JournalEntryLineItem.id, net_balance) \ - .join(Account) \ - .join(offset, be(JournalEntryLineItem.id - == offset.c.original_line_item_id), - isouter=True) \ - .filter(be(Account.id == account.id), - sa.or_(sa.and_(Account.base_code.startswith("2"), - sa.not_(JournalEntryLineItem.is_debit)), - sa.and_(Account.base_code.startswith("1"), - JournalEntryLineItem.is_debit))) \ - .group_by(JournalEntryLineItem.id) \ - .having(sa.or_(sa.func.count(offset.c.id) == 0, net_balance != 0)) - net_balances: dict[int, Decimal] \ - = {x.id: x.net_balance - for x in db.session.execute(select_net_balances).all()} - line_items: list[JournalEntryLineItem] = JournalEntryLineItem.query \ - .filter(JournalEntryLineItem.id.in_({x for x in net_balances})) \ - .join(JournalEntry) \ - .order_by(JournalEntry.date, JournalEntry.no, - JournalEntryLineItem.is_debit, JournalEntryLineItem.no) \ - .options(selectinload(JournalEntryLineItem.currency), - selectinload(JournalEntryLineItem.journal_entry)).all() - for line_item in line_items: - line_item.net_balance = line_item.amount \ - if net_balances[line_item.id] is None \ - else net_balances[line_item.id] - return line_items