From 052b62cdd46f799ec9434b1af1d1362645935a2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Sat, 8 Apr 2023 10:47:50 +0800 Subject: [PATCH] Moved the __query_line_items method in the UnappliedOriginalLineItems report to the new "accounting.utils.unapplied" module, to share this query. --- src/accounting/report/reports/unapplied.py | 52 +--------------- src/accounting/utils/unapplied.py | 72 ++++++++++++++++++++++ 2 files changed, 75 insertions(+), 49 deletions(-) create mode 100644 src/accounting/utils/unapplied.py diff --git a/src/accounting/report/reports/unapplied.py b/src/accounting/report/reports/unapplied.py index 0b9a029..56dcbc8 100644 --- a/src/accounting/report/reports/unapplied.py +++ b/src/accounting/report/reports/unapplied.py @@ -20,14 +20,10 @@ 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 Account, JournalEntry, \ - JournalEntryLineItem +from accounting.models import Account, JournalEntryLineItem from accounting.report.utils.base_page_params import BasePageParams from accounting.report.utils.base_report import BaseReport from accounting.report.utils.csv_export import BaseCSVRow, csv_download @@ -36,9 +32,8 @@ 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.urls import unapplied_url -from accounting.utils.cast import be -from accounting.utils.offset_alias import offset_alias from accounting.utils.pagination import Pagination +from accounting.utils.unapplied import get_unapplied_original_line_items class CSVRow(BaseCSVRow): @@ -154,50 +149,9 @@ class UnappliedOriginalLineItems(BaseReport): self.__account: Account = account """The account.""" self.__line_items: list[JournalEntryLineItem] \ - = self.__query_line_items() + = get_unapplied_original_line_items(self.__account) """The line items.""" - def __query_line_items(self) -> list[JournalEntryLineItem]: - """Queries and returns the line items. - - :return: The line items. - """ - 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 csv(self) -> Response: """Returns the report as CSV for download. diff --git a/src/accounting/utils/unapplied.py b/src/accounting/utils/unapplied.py new file mode 100644 index 0000000..b6cf0b7 --- /dev/null +++ b/src/accounting/utils/unapplied.py @@ -0,0 +1,72 @@ +# 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