Removed the "be" cast function to cast data type for the binary expressions. It is to be replaced by the mapped type hints.

This commit is contained in:
依瑪貓 2023-04-23 10:35:50 +08:00
parent 09dd5ae541
commit eed4c923f6
9 changed files with 26 additions and 52 deletions

View File

@ -30,7 +30,6 @@ from accounting import db
from accounting.forms import CurrencyExists from accounting.forms import CurrencyExists
from accounting.locale import lazy_gettext from accounting.locale import lazy_gettext
from accounting.models import JournalEntryLineItem from accounting.models import JournalEntryLineItem
from accounting.utils.cast import be
from accounting.utils.offset_alias import offset_alias from accounting.utils.offset_alias import offset_alias
from accounting.utils.strip_text import strip_text from accounting.utils.strip_text import strip_text
from .line_item import LineItemForm, CreditLineItemForm, DebitLineItemForm from .line_item import LineItemForm, CreditLineItemForm, DebitLineItemForm
@ -75,8 +74,8 @@ class KeepCurrencyWhenHavingOffset:
offset: sa.Alias = offset_alias() offset: sa.Alias = offset_alias()
original_line_items: list[JournalEntryLineItem]\ original_line_items: list[JournalEntryLineItem]\
= JournalEntryLineItem.query\ = JournalEntryLineItem.query\
.join(offset, be(JournalEntryLineItem.id .join(offset,
== offset.c.original_line_item_id), JournalEntryLineItem.id == offset.c.original_line_item_id,
isouter=True)\ isouter=True)\
.filter(JournalEntryLineItem.id .filter(JournalEntryLineItem.id
.in_({x.id.data for x in form.line_items .in_({x.id.data for x in form.line_items

View File

@ -33,7 +33,6 @@ from accounting.forms import ACCOUNT_REQUIRED, AccountExists, IsDebitAccount, \
from accounting.locale import lazy_gettext from accounting.locale import lazy_gettext
from accounting.models import Account, JournalEntry, JournalEntryLineItem from accounting.models import Account, JournalEntry, JournalEntryLineItem
from accounting.template_filters import format_amount from accounting.template_filters import format_amount
from accounting.utils.cast import be
from accounting.utils.random_id import new_id from accounting.utils.random_id import new_id
from accounting.utils.strip_text import strip_text from accounting.utils.strip_text import strip_text
from accounting.utils.user import get_current_user_pk from accounting.utils.user import get_current_user_pk
@ -198,13 +197,13 @@ class NotExceedingOriginalLineItemNetBalance:
existing_line_item_id \ existing_line_item_id \
= {x.id for x in form.journal_entry_form.obj.line_items} = {x.id for x in form.journal_entry_form.obj.line_items}
offset_total_func: sa.Function = sa.func.sum(sa.case( offset_total_func: sa.Function = sa.func.sum(sa.case(
(be(JournalEntryLineItem.is_debit == is_debit), (JournalEntryLineItem.is_debit == is_debit,
JournalEntryLineItem.amount), JournalEntryLineItem.amount),
else_=-JournalEntryLineItem.amount)) else_=-JournalEntryLineItem.amount))
offset_total_but_form: Decimal | None = db.session.scalar( offset_total_but_form: Decimal | None = db.session.scalar(
sa.select(offset_total_func) sa.select(offset_total_func)
.filter(be(JournalEntryLineItem.original_line_item_id .filter(JournalEntryLineItem.original_line_item_id
== original_line_item.id), == original_line_item.id,
JournalEntryLineItem.id.not_in(existing_line_item_id))) JournalEntryLineItem.id.not_in(existing_line_item_id)))
if offset_total_but_form is None: if offset_total_but_form is None:
offset_total_but_form = Decimal("0") offset_total_but_form = Decimal("0")
@ -232,8 +231,7 @@ class NotLessThanOffsetTotal:
(JournalEntryLineItem.is_debit != is_debit, (JournalEntryLineItem.is_debit != is_debit,
JournalEntryLineItem.amount), JournalEntryLineItem.amount),
else_=-JournalEntryLineItem.amount)))\ else_=-JournalEntryLineItem.amount)))\
.filter(be(JournalEntryLineItem.original_line_item_id .filter(JournalEntryLineItem.original_line_item_id == form.id.data)
== form.id.data))
offset_total: Decimal | None = db.session.scalar(select_offset_total) offset_total: Decimal | None = db.session.scalar(select_offset_total)
if offset_total is not None and field.data < offset_total: if offset_total is not None and field.data < offset_total:
raise ValidationError(lazy_gettext( raise ValidationError(lazy_gettext(

View File

@ -24,7 +24,6 @@ from sqlalchemy.orm import selectinload
from accounting import db from accounting import db
from accounting.models import Account, JournalEntry, JournalEntryLineItem from accounting.models import Account, JournalEntry, JournalEntryLineItem
from accounting.utils.cast import be
from accounting.utils.offset_alias import offset_alias from accounting.utils.offset_alias import offset_alias
@ -45,8 +44,7 @@ def get_selectable_original_line_items(
offset: sa.Alias = offset_alias() offset: sa.Alias = offset_alias()
net_balance: sa.Label = (JournalEntryLineItem.amount + sa.func.sum(sa.case( net_balance: sa.Label = (JournalEntryLineItem.amount + sa.func.sum(sa.case(
(offset.c.id.in_(line_item_id_on_form), 0), (offset.c.id.in_(line_item_id_on_form), 0),
(be(offset.c.is_debit == JournalEntryLineItem.is_debit), (offset.c.is_debit == JournalEntryLineItem.is_debit, offset.c.amount),
offset.c.amount),
else_=-offset.c.amount))).label("net_balance") else_=-offset.c.amount))).label("net_balance")
conditions: list[sa.BinaryExpression] = [Account.is_need_offset] conditions: list[sa.BinaryExpression] = [Account.is_need_offset]
sub_conditions: list[sa.BinaryExpression] = [] sub_conditions: list[sa.BinaryExpression] = []
@ -60,8 +58,8 @@ def get_selectable_original_line_items(
select_net_balances: sa.Select \ select_net_balances: sa.Select \
= sa.select(JournalEntryLineItem.id, net_balance)\ = sa.select(JournalEntryLineItem.id, net_balance)\
.join(Account)\ .join(Account)\
.join(offset, be(JournalEntryLineItem.id .join(offset,
== offset.c.original_line_item_id), JournalEntryLineItem.id == offset.c.original_line_item_id,
isouter=True)\ isouter=True)\
.filter(*conditions)\ .filter(*conditions)\
.group_by(JournalEntryLineItem.id)\ .group_by(JournalEntryLineItem.id)\

View File

@ -37,7 +37,6 @@ from accounting.report.utils.option_link import OptionLink
from accounting.report.utils.report_chooser import ReportChooser 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.urls import income_expenses_url from accounting.report.utils.urls import income_expenses_url
from accounting.utils.cast import be
from accounting.utils.current_account import CurrentAccount from accounting.utils.current_account import CurrentAccount
from accounting.utils.pagination import Pagination from accounting.utils.pagination import Pagination
@ -122,8 +121,7 @@ class LineItemCollector:
else_=-JournalEntryLineItem.amount)) else_=-JournalEntryLineItem.amount))
select: sa.Select = sa.Select(balance_func)\ select: sa.Select = sa.Select(balance_func)\
.join(JournalEntry).join(Account)\ .join(JournalEntry).join(Account)\
.filter(be(JournalEntryLineItem.currency_code .filter(JournalEntryLineItem.currency_code == self.__currency.code,
== self.__currency.code),
self.__account_condition, self.__account_condition,
JournalEntry.date < self.__period.start) JournalEntry.date < self.__period.start)
balance: int | None = db.session.scalar(select) balance: int | None = db.session.scalar(select)
@ -347,8 +345,7 @@ class PageParams(BasePageParams):
self.account.id == 0)] self.account.id == 0)]
in_use: sa.Select = sa.Select(JournalEntryLineItem.account_id)\ in_use: sa.Select = sa.Select(JournalEntryLineItem.account_id)\
.join(Account)\ .join(Account)\
.filter(be(JournalEntryLineItem.currency_code .filter(JournalEntryLineItem.currency_code == self.currency.code,
== self.currency.code),
CurrentAccount.sql_condition())\ CurrentAccount.sql_condition())\
.group_by(JournalEntryLineItem.account_id) .group_by(JournalEntryLineItem.account_id)
options.extend([OptionLink(str(x), options.extend([OptionLink(str(x),

View File

@ -37,7 +37,6 @@ from accounting.report.utils.option_link import OptionLink
from accounting.report.utils.report_chooser import ReportChooser 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.urls import ledger_url from accounting.report.utils.urls import ledger_url
from accounting.utils.cast import be
from accounting.utils.pagination import Pagination from accounting.utils.pagination import Pagination
@ -118,10 +117,8 @@ class LineItemCollector:
(JournalEntryLineItem.is_debit, JournalEntryLineItem.amount), (JournalEntryLineItem.is_debit, JournalEntryLineItem.amount),
else_=-JournalEntryLineItem.amount)) else_=-JournalEntryLineItem.amount))
select: sa.Select = sa.Select(balance_func).join(JournalEntry)\ select: sa.Select = sa.Select(balance_func).join(JournalEntry)\
.filter(be(JournalEntryLineItem.currency_code .filter(JournalEntryLineItem.currency_code == self.__currency.code,
== self.__currency.code), JournalEntryLineItem.account_id == self.__account.id,
be(JournalEntryLineItem.account_id
== self.__account.id),
JournalEntry.date < self.__period.start) JournalEntry.date < self.__period.start)
balance: int | None = db.session.scalar(select) balance: int | None = db.session.scalar(select)
if balance is None: if balance is None:
@ -313,8 +310,7 @@ class PageParams(BasePageParams):
:return: The account options. :return: The account options.
""" """
in_use: sa.Select = sa.Select(JournalEntryLineItem.account_id)\ in_use: sa.Select = sa.Select(JournalEntryLineItem.account_id)\
.filter(be(JournalEntryLineItem.currency_code .filter(JournalEntryLineItem.currency_code == self.currency.code)\
== self.currency.code))\
.group_by(JournalEntryLineItem.account_id) .group_by(JournalEntryLineItem.account_id)
return [OptionLink(str(x), ledger_url(self.currency, x, self.period), return [OptionLink(str(x), ledger_url(self.currency, x, self.period),
x.id == self.account.id) x.id == self.account.id)

View File

@ -32,7 +32,6 @@ from accounting.report.utils.base_report import BaseReport
from accounting.report.utils.csv_export import csv_download from accounting.report.utils.csv_export import csv_download
from accounting.report.utils.report_chooser import ReportChooser 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.utils.cast import be
from accounting.utils.pagination import Pagination from accounting.utils.pagination import Pagination
from accounting.utils.query import parse_query_keywords from accounting.utils.query import parse_query_keywords
from .journal import get_csv_rows from .journal import get_csv_rows
@ -128,9 +127,8 @@ class LineItemCollector:
journal_entry_date: datetime journal_entry_date: datetime
try: try:
journal_entry_date = datetime.strptime(k, "%Y") journal_entry_date = datetime.strptime(k, "%Y")
conditions.append( conditions.append(sa.extract("year", JournalEntry.date)
be(sa.extract("year", JournalEntry.date) == journal_entry_date.year)
== journal_entry_date.year))
except ValueError: except ValueError:
pass pass
try: try:

View File

@ -24,7 +24,6 @@ import sqlalchemy as sa
from accounting import db from accounting import db
from accounting.models import Currency, Account, JournalEntry, \ from accounting.models import Currency, Account, JournalEntry, \
JournalEntryLineItem JournalEntryLineItem
from accounting.utils.cast import be
from accounting.utils.offset_alias import offset_alias from accounting.utils.offset_alias import offset_alias
@ -38,17 +37,17 @@ def get_accounts_with_unapplied(currency: Currency) -> list[Account]:
net_balance: sa.Label \ net_balance: sa.Label \
= (JournalEntryLineItem.amount = (JournalEntryLineItem.amount
+ sa.func.sum(sa.case( + sa.func.sum(sa.case(
(be(offset.c.is_debit == JournalEntryLineItem.is_debit), (offset.c.is_debit == JournalEntryLineItem.is_debit,
offset.c.amount), offset.c.amount),
else_=-offset.c.amount))).label("net_balance") else_=-offset.c.amount))).label("net_balance")
select_unapplied: sa.Select \ select_unapplied: sa.Select \
= sa.select(JournalEntryLineItem.id)\ = sa.select(JournalEntryLineItem.id)\
.join(JournalEntry).join(Account)\ .join(JournalEntry).join(Account)\
.join(offset, be(JournalEntryLineItem.id .join(offset,
== offset.c.original_line_item_id), JournalEntryLineItem.id == offset.c.original_line_item_id,
isouter=True)\ isouter=True)\
.filter(Account.is_need_offset, .filter(Account.is_need_offset,
be(JournalEntryLineItem.currency_code == currency.code), JournalEntryLineItem.currency_code == currency.code,
sa.or_(sa.and_(Account.base_code.startswith("2"), sa.or_(sa.and_(Account.base_code.startswith("2"),
sa.not_(JournalEntryLineItem.is_debit)), sa.not_(JournalEntryLineItem.is_debit)),
sa.and_(Account.base_code.startswith("1"), sa.and_(Account.base_code.startswith("1"),
@ -84,17 +83,17 @@ def get_net_balances(currency: Currency, account: Account) \
net_balance: sa.Label \ net_balance: sa.Label \
= (JournalEntryLineItem.amount = (JournalEntryLineItem.amount
+ sa.func.sum(sa.case( + sa.func.sum(sa.case(
(be(offset.c.is_debit == JournalEntryLineItem.is_debit), (offset.c.is_debit == JournalEntryLineItem.is_debit,
offset.c.amount), offset.c.amount),
else_=-offset.c.amount))).label("net_balance") else_=-offset.c.amount))).label("net_balance")
select_net_balances: sa.Select \ select_net_balances: sa.Select \
= sa.select(JournalEntryLineItem.id, net_balance) \ = sa.select(JournalEntryLineItem.id, net_balance) \
.join(JournalEntry).join(Account) \ .join(JournalEntry).join(Account) \
.join(offset, be(JournalEntryLineItem.id .join(offset,
== offset.c.original_line_item_id), JournalEntryLineItem.id == offset.c.original_line_item_id,
isouter=True) \ isouter=True) \
.filter(be(Account.id == account.id), .filter(Account.id == account.id,
be(JournalEntryLineItem.currency_code == currency.code), JournalEntryLineItem.currency_code == currency.code,
sa.or_(sa.and_(Account.base_code.startswith("2"), sa.or_(sa.and_(Account.base_code.startswith("2"),
sa.not_(JournalEntryLineItem.is_debit)), sa.not_(JournalEntryLineItem.is_debit)),
sa.and_(Account.base_code.startswith("1"), sa.and_(Account.base_code.startswith("1"),

View File

@ -22,7 +22,6 @@ import sqlalchemy as sa
from accounting import db from accounting import db
from accounting.models import Currency, Account, JournalEntry, \ from accounting.models import Currency, Account, JournalEntry, \
JournalEntryLineItem JournalEntryLineItem
from accounting.utils.cast import be
def get_accounts_with_unmatched(currency: Currency) -> list[Account]: def get_accounts_with_unmatched(currency: Currency) -> list[Account]:
@ -38,7 +37,7 @@ def get_accounts_with_unmatched(currency: Currency) -> list[Account]:
.select_from(Account)\ .select_from(Account)\
.join(JournalEntryLineItem, isouter=True).join(JournalEntry)\ .join(JournalEntryLineItem, isouter=True).join(JournalEntry)\
.filter(Account.is_need_offset, .filter(Account.is_need_offset,
be(JournalEntryLineItem.currency_code == currency.code), JournalEntryLineItem.currency_code == currency.code,
JournalEntryLineItem.original_line_item_id.is_(None), JournalEntryLineItem.original_line_item_id.is_(None),
sa.or_(sa.and_(Account.base_code.startswith("2"), sa.or_(sa.and_(Account.base_code.startswith("2"),
JournalEntryLineItem.is_debit), JournalEntryLineItem.is_debit),

View File

@ -25,16 +25,6 @@ import typing as t
import sqlalchemy as sa import sqlalchemy as sa
def be(expression: t.Any) -> sa.BinaryExpression:
"""Casts the SQLAlchemy binary expression to the binary expression type.
:param expression: The binary expression.
:return: The binary expression itself.
"""
assert isinstance(expression, sa.BinaryExpression)
return expression
def s(message: t.Any) -> str: def s(message: t.Any) -> str:
"""Casts the LazyString message to the string type. """Casts the LazyString message to the string type.