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

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

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.report.utils.unapplied import get_net_balances
-from accounting.utils.offset_alias import offset_alias
 
 
 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.
 

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()}