From 412da170e1958fae8171bce0101c0ae3c630cf03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Tue, 18 Apr 2023 07:04:00 +0800 Subject: [PATCH] Added the get_net_balances function to the "accounting.report.utils.unapplied" module to replace the __get_net_balances methods of the OffsetMatcher and UnappliedOriginalLineItems classes. --- src/accounting/report/reports/unapplied.py | 45 ++----------------- src/accounting/report/utils/offset_matcher.py | 37 ++------------- src/accounting/report/utils/unapplied.py | 43 ++++++++++++++++++ 3 files changed, 50 insertions(+), 75 deletions(-) diff --git a/src/accounting/report/reports/unapplied.py b/src/accounting/report/reports/unapplied.py index d24ec25..5b2c94d 100644 --- a/src/accounting/report/reports/unapplied.py +++ b/src/accounting/report/reports/unapplied.py @@ -20,11 +20,9 @@ from datetime import date from decimal import Decimal -import sqlalchemy as sa from flask import render_template, Response from sqlalchemy.orm import selectinload -from accounting import db from accounting.locale import gettext from accounting.models import Currency, Account, JournalEntry, \ JournalEntryLineItem @@ -36,10 +34,9 @@ from accounting.report.utils.csv_export import BaseCSVRow, csv_download, \ from accounting.report.utils.option_link import OptionLink from accounting.report.utils.report_chooser import ReportChooser from accounting.report.utils.report_type import ReportType -from accounting.report.utils.unapplied import get_accounts_with_unapplied +from accounting.report.utils.unapplied import get_accounts_with_unapplied, \ + get_net_balances from accounting.report.utils.urls import unapplied_url -from accounting.utils.cast import be -from accounting.utils.offset_alias import offset_alias from accounting.utils.pagination import Pagination @@ -193,7 +190,8 @@ class UnappliedOriginalLineItems(BaseReport): :return: The line items. """ - net_balances: dict[int, Decimal | None] = self.__get_net_balances() + net_balances: dict[int, Decimal | None] \ + = get_net_balances(self.__currency, self.__account, self.__period) line_items: list[JournalEntryLineItem] = JournalEntryLineItem.query \ .join(Account).join(JournalEntry) \ .filter(JournalEntryLineItem.id.in_(net_balances)) \ @@ -207,41 +205,6 @@ class UnappliedOriginalLineItems(BaseReport): else net_balances[line_item.id] return line_items - def __get_net_balances(self) -> dict[int, Decimal | None]: - """Returns the net balances of the unapplied line items of the account. - - :return: The net balances of the unapplied 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") - conditions: list[sa.BinaryExpression] \ - = [be(Account.id == self.__account.id), - be(JournalEntryLineItem.currency_code == self.__currency.code), - sa.or_(sa.and_(Account.base_code.startswith("2"), - sa.not_(JournalEntryLineItem.is_debit)), - sa.and_(Account.base_code.startswith("1"), - JournalEntryLineItem.is_debit))] - if self.__period.start is not None: - conditions.append(JournalEntry.date >= self.__period.start) - if self.__period.end is not None: - conditions.append(JournalEntry.date <= self.__period.end) - select_net_balances: sa.Select \ - = sa.select(JournalEntryLineItem.id, net_balance) \ - .join(JournalEntry).join(Account) \ - .join(offset, be(JournalEntryLineItem.id - == offset.c.original_line_item_id), - isouter=True) \ - .filter(*conditions) \ - .group_by(JournalEntryLineItem.id) \ - .having(sa.or_(sa.func.count(offset.c.id) == 0, net_balance != 0)) - return {x.id: x.net_balance - for x in db.session.execute(select_net_balances).all()} - def csv(self) -> Response: """Returns the report as CSV for download. diff --git a/src/accounting/report/utils/offset_matcher.py b/src/accounting/report/utils/offset_matcher.py index 80b187b..8439f44 100644 --- a/src/accounting/report/utils/offset_matcher.py +++ b/src/accounting/report/utils/offset_matcher.py @@ -23,13 +23,11 @@ import sqlalchemy as sa from flask_babel import LazyString from sqlalchemy.orm import selectinload -from accounting import db from accounting.locale import lazy_gettext from accounting.models import Currency, Account, JournalEntry, \ JournalEntryLineItem from accounting.report.period import Period -from accounting.utils.cast import be -from accounting.utils.offset_alias import offset_alias +from accounting.report.utils.unapplied import get_net_balances class OffsetPair: @@ -118,7 +116,8 @@ class OffsetMatcher: :return: The unapplied original line items and unmatched offsets of the account. """ - net_balances: dict[int, Decimal | None] = self.__get_net_balances() + net_balances: dict[int, Decimal | None] \ + = get_net_balances(self.__currency, self.__account, None) unmatched_offset_condition: sa.BinaryExpression \ = sa.and_(Account.id == self.__account.id, JournalEntryLineItem.currency_code @@ -148,36 +147,6 @@ class OffsetMatcher: if not x.is_offset] self.__populate_accumulated_balances() - def __get_net_balances(self) -> dict[int, Decimal | None]: - """Returns the net balances of the unapplied line items of the account. - - :return: The net balances of the unapplied 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), - be(JournalEntryLineItem.currency_code - == self.__currency.code), - 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)) - return {x.id: x.net_balance - for x in db.session.execute(select_net_balances).all()} - def __populate_accumulated_balances(self) -> None: """Populates the accumulated balances of the line items. diff --git a/src/accounting/report/utils/unapplied.py b/src/accounting/report/utils/unapplied.py index 75ccaa6..6edc51e 100644 --- a/src/accounting/report/utils/unapplied.py +++ b/src/accounting/report/utils/unapplied.py @@ -17,6 +17,8 @@ """The unapplied original line item utilities. """ +from decimal import Decimal + import sqlalchemy as sa from accounting import db @@ -77,3 +79,44 @@ def get_accounts_with_unapplied(currency: Currency, for account in accounts: account.count = counts[account.id] return accounts + + +def get_net_balances(currency: Currency, account: Account, + period: Period | None) -> dict[int, Decimal | None]: + """Returns the net balances of the unapplied line items of the account. + + :param currency: The currency. + :param account: The account. + :param period: The period, or None for all time. + :return: The net balances of the unapplied 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") + conditions: list[sa.BinaryExpression] \ + = [be(Account.id == account.id), + be(JournalEntryLineItem.currency_code == currency.code), + sa.or_(sa.and_(Account.base_code.startswith("2"), + sa.not_(JournalEntryLineItem.is_debit)), + sa.and_(Account.base_code.startswith("1"), + JournalEntryLineItem.is_debit))] + if period is not None: + if period.start is not None: + conditions.append(be(JournalEntry.date >= period.start)) + if period.end is not None: + conditions.append(be(JournalEntry.date <= period.end)) + select_net_balances: sa.Select \ + = sa.select(JournalEntryLineItem.id, net_balance) \ + .join(JournalEntry).join(Account) \ + .join(offset, be(JournalEntryLineItem.id + == offset.c.original_line_item_id), + isouter=True) \ + .filter(*conditions) \ + .group_by(JournalEntryLineItem.id) \ + .having(sa.or_(sa.func.count(offset.c.id) == 0, net_balance != 0)) + return {x.id: x.net_balance + for x in db.session.execute(select_net_balances).all()}