Moved the __query_line_items method in the UnappliedOriginalLineItems report to the new "accounting.utils.unapplied" module, to share this query.

This commit is contained in:
依瑪貓 2023-04-08 10:47:50 +08:00
parent 3728a4037d
commit 052b62cdd4
2 changed files with 75 additions and 49 deletions

View File

@ -20,14 +20,10 @@
from datetime import date from datetime import date
from decimal import Decimal from decimal import Decimal
import sqlalchemy as sa
from flask import render_template, Response from flask import render_template, Response
from sqlalchemy.orm import selectinload
from accounting import db
from accounting.locale import gettext from accounting.locale import gettext
from accounting.models import Account, JournalEntry, \ from accounting.models import Account, JournalEntryLineItem
JournalEntryLineItem
from accounting.report.utils.base_page_params import BasePageParams from accounting.report.utils.base_page_params import BasePageParams
from accounting.report.utils.base_report import BaseReport from accounting.report.utils.base_report import BaseReport
from accounting.report.utils.csv_export import BaseCSVRow, csv_download 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.report_type import ReportType
from accounting.report.utils.unapplied import get_accounts_with_unapplied from accounting.report.utils.unapplied import get_accounts_with_unapplied
from accounting.report.utils.urls import unapplied_url 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.pagination import Pagination
from accounting.utils.unapplied import get_unapplied_original_line_items
class CSVRow(BaseCSVRow): class CSVRow(BaseCSVRow):
@ -154,50 +149,9 @@ class UnappliedOriginalLineItems(BaseReport):
self.__account: Account = account self.__account: Account = account
"""The account.""" """The account."""
self.__line_items: list[JournalEntryLineItem] \ self.__line_items: list[JournalEntryLineItem] \
= self.__query_line_items() = get_unapplied_original_line_items(self.__account)
"""The line items.""" """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: def csv(self) -> Response:
"""Returns the report as CSV for download. """Returns the report as CSV for download.

View File

@ -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