mia-accounting/src/accounting/models.py

807 lines
30 KiB
Python

# The Mia! Accounting Project.
# Author: imacat@mail.imacat.idv.tw (imacat), 2023/1/25
# 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 data models.
"""
from __future__ import annotations
import re
import typing as t
from decimal import Decimal
import sqlalchemy as sa
from babel import Locale
from flask_babel import get_locale, get_babel
from sqlalchemy import text
from accounting import db
from accounting.locale import gettext
from accounting.utils.user import user_cls, user_pk_column
class BaseAccount(db.Model):
"""A base account."""
__tablename__ = "accounting_base_accounts"
"""The table name."""
code = db.Column(db.String, nullable=False, primary_key=True)
"""The code."""
title_l10n = db.Column("title", db.String, nullable=False)
"""The title."""
l10n = db.relationship("BaseAccountL10n", back_populates="account",
lazy=False)
"""The localized titles."""
accounts = db.relationship("Account", back_populates="base")
"""The descendant accounts under the base account."""
def __str__(self) -> str:
"""Returns the string representation of the base account.
:return: The string representation of the base account.
"""
return f"{self.code} {self.title.title()}"
@property
def title(self) -> str:
"""Returns the title in the current locale.
:return: The title in the current locale.
"""
current_locale: Locale = get_locale()
if current_locale == get_babel().instance.default_locale:
return self.title_l10n
for l10n in self.l10n:
if l10n.locale == str(current_locale):
return l10n.title
return self.title_l10n
@property
def query_values(self) -> list[str]:
"""Returns the values to be queried.
:return: The values to be queried.
"""
return [self.code, self.title_l10n] + [x.title for x in self.l10n]
class BaseAccountL10n(db.Model):
"""A localized base account title."""
__tablename__ = "accounting_base_accounts_l10n"
"""The table name."""
account_code = db.Column(db.String,
db.ForeignKey(BaseAccount.code,
onupdate="CASCADE",
ondelete="CASCADE"),
nullable=False, primary_key=True)
"""The code of the account."""
account = db.relationship(BaseAccount, back_populates="l10n")
"""The account."""
locale = db.Column(db.String, nullable=False, primary_key=True)
"""The locale."""
title = db.Column(db.String, nullable=False)
"""The localized title."""
class Account(db.Model):
"""An account."""
__tablename__ = "accounting_accounts"
"""The table name."""
id = db.Column(db.Integer, nullable=False, primary_key=True,
autoincrement=False)
"""The account ID."""
base_code = db.Column(db.String,
db.ForeignKey(BaseAccount.code, onupdate="CASCADE",
ondelete="CASCADE"),
nullable=False)
"""The code of the base account."""
base = db.relationship(BaseAccount, back_populates="accounts")
"""The base account."""
no = db.Column(db.Integer, nullable=False, default=text("1"))
"""The account number under the base account."""
title_l10n = db.Column("title", db.String, nullable=False)
"""The title."""
is_need_offset = db.Column(db.Boolean, nullable=False, default=False)
"""Whether the journal entry line items of this account need offset."""
created_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of creation."""
created_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the creator."""
created_by = db.relationship(user_cls, foreign_keys=created_by_id)
"""The creator."""
updated_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of last update."""
updated_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the updator."""
updated_by = db.relationship(user_cls, foreign_keys=updated_by_id)
"""The updator."""
l10n = db.relationship("AccountL10n", back_populates="account",
lazy=False)
"""The localized titles."""
line_items = db.relationship("JournalEntryLineItem",
back_populates="account")
"""The journal entry line items."""
CASH_CODE: str = "1111-001"
"""The code of the cash account,"""
ACCUMULATED_CHANGE_CODE: str = "3351-001"
"""The code of the accumulated-change account,"""
NET_CHANGE_CODE: str = "3353-001"
"""The code of the net-change account,"""
def __str__(self) -> str:
"""Returns the string representation of this account.
:return: The string representation of this account.
"""
return f"{self.base_code}-{self.no:03d} {self.title.title()}"
@property
def code(self) -> str:
"""Returns the code.
:return: The code.
"""
return f"{self.base_code}-{self.no:03d}"
@property
def title(self) -> str:
"""Returns the title in the current locale.
:return: The title in the current locale.
"""
current_locale: Locale = get_locale()
if current_locale == get_babel().instance.default_locale:
return self.title_l10n
for l10n in self.l10n:
if l10n.locale == str(current_locale):
return l10n.title
return self.title_l10n
@title.setter
def title(self, value: str) -> None:
"""Sets the title in the current locale.
:param value: The new title.
:return: None.
"""
if self.title_l10n is None:
self.title_l10n = value
return
current_locale: Locale = get_locale()
if current_locale == get_babel().instance.default_locale:
self.title_l10n = value
return
for l10n in self.l10n:
if l10n.locale == str(current_locale):
l10n.title = value
return
self.l10n.append(AccountL10n(locale=str(current_locale), title=value))
@property
def is_real(self) -> bool:
"""Returns whether the account is a real account.
:return: True if the account is a real account, or False otherwise.
"""
return self.base_code[0] in {"1", "2", "3"}
@property
def is_nominal(self) -> bool:
"""Returns whether the account is a nominal account.
:return: True if the account is a nominal account, or False otherwise.
"""
return not self.is_real
@property
def query_values(self) -> list[str]:
"""Returns the values to be queried.
:return: The values to be queried.
"""
return [self.code, self.title_l10n] + [x.title for x in self.l10n]
@property
def is_modified(self) -> bool:
"""Returns whether a product account was modified.
:return: True if modified, or False otherwise.
"""
if db.session.is_modified(self):
return True
for l10n in self.l10n:
if db.session.is_modified(l10n):
return True
return False
@property
def can_delete(self) -> bool:
"""Returns whether the account can be deleted.
:return: True if the account can be deleted, or False otherwise.
"""
if self.code in {"1111-001", "3351-001", "3353-001"}:
return False
return len(self.line_items) == 0
def delete(self) -> None:
"""Deletes this account.
:return: None.
"""
AccountL10n.query.filter(AccountL10n.account == self).delete()
cls: t.Type[t.Self] = self.__class__
cls.query.filter(cls.id == self.id).delete()
@classmethod
def find_by_code(cls, code: str) -> t.Self | None:
"""Finds an account by its code.
:param code: The code.
:return: The account, or None if this account does not exist.
"""
m = re.match(r"^([1-9]{4})-(\d{3})$", code)
if m is None:
return None
return cls.query.filter(cls.base_code == m.group(1),
cls.no == int(m.group(2))).first()
@classmethod
def selectable_debit(cls) -> list[t.Self]:
"""Returns the selectable debit accounts.
Payable line items can not start from debit.
:return: The selectable debit accounts.
"""
return cls.query.filter(sa.or_(cls.base_code.startswith("1"),
sa.and_(cls.base_code.startswith("2"),
sa.not_(cls.is_need_offset)),
cls.base_code.startswith("3"),
cls.base_code.startswith("5"),
cls.base_code.startswith("6"),
cls.base_code.startswith("75"),
cls.base_code.startswith("76"),
cls.base_code.startswith("77"),
cls.base_code.startswith("78"),
cls.base_code.startswith("8"),
cls.base_code.startswith("9")),
cls.base_code != "3351",
cls.base_code != "3353")\
.order_by(cls.base_code, cls.no).all()
@classmethod
def selectable_credit(cls) -> list[t.Self]:
"""Returns the selectable debit accounts.
Receivable line items can not start from credit.
:return: The selectable debit accounts.
"""
return cls.query.filter(sa.or_(sa.and_(cls.base_code.startswith("1"),
sa.not_(cls.is_need_offset)),
cls.base_code.startswith("2"),
cls.base_code.startswith("3"),
cls.base_code.startswith("4"),
cls.base_code.startswith("71"),
cls.base_code.startswith("72"),
cls.base_code.startswith("73"),
cls.base_code.startswith("74"),
cls.base_code.startswith("8"),
cls.base_code.startswith("9")),
cls.base_code != "3351",
cls.base_code != "3353")\
.order_by(cls.base_code, cls.no).all()
@classmethod
def cash(cls) -> t.Self:
"""Returns the cash account.
:return: The cash account
"""
return cls.find_by_code(cls.CASH_CODE)
@classmethod
def accumulated_change(cls) -> t.Self:
"""Returns the accumulated-change account.
:return: The accumulated-change account
"""
return cls.find_by_code(cls.ACCUMULATED_CHANGE_CODE)
class AccountL10n(db.Model):
"""A localized account title."""
__tablename__ = "accounting_accounts_l10n"
"""The table name."""
account_id = db.Column(db.Integer,
db.ForeignKey(Account.id, onupdate="CASCADE",
ondelete="CASCADE"),
nullable=False, primary_key=True)
"""The account ID."""
account = db.relationship(Account, back_populates="l10n")
"""The account."""
locale = db.Column(db.String, nullable=False, primary_key=True)
"""The locale."""
title = db.Column(db.String, nullable=False)
"""The localized title."""
class Currency(db.Model):
"""A currency."""
__tablename__ = "accounting_currencies"
"""The table name."""
code = db.Column(db.String, nullable=False, primary_key=True)
"""The code."""
name_l10n = db.Column("name", db.String, nullable=False)
"""The name."""
created_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of creation."""
created_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the creator."""
created_by = db.relationship(user_cls, foreign_keys=created_by_id)
"""The creator."""
updated_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of last update."""
updated_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the updator."""
updated_by = db.relationship(user_cls, foreign_keys=updated_by_id)
"""The updator."""
l10n = db.relationship("CurrencyL10n", back_populates="currency",
lazy=False)
"""The localized names."""
line_items = db.relationship("JournalEntryLineItem",
back_populates="currency")
"""The journal entry line items."""
def __str__(self) -> str:
"""Returns the string representation of the currency.
:return: The string representation of the currency.
"""
return f"{self.name.title()} ({self.code})"
@property
def name(self) -> str:
"""Returns the name in the current locale.
:return: The name in the current locale.
"""
current_locale: Locale = get_locale()
if current_locale == get_babel().instance.default_locale:
return self.name_l10n
for l10n in self.l10n:
if l10n.locale == str(current_locale):
return l10n.name
return self.name_l10n
@name.setter
def name(self, value: str) -> None:
"""Sets the name in the current locale.
:param value: The new name.
:return: None.
"""
if self.name_l10n is None:
self.name_l10n = value
return
current_locale: Locale = get_locale()
if current_locale == get_babel().instance.default_locale:
self.name_l10n = value
return
for l10n in self.l10n:
if l10n.locale == str(current_locale):
l10n.name = value
return
self.l10n.append(CurrencyL10n(locale=str(current_locale), name=value))
@property
def is_modified(self) -> bool:
"""Returns whether a product account was modified.
:return: True if modified, or False otherwise.
"""
if db.session.is_modified(self):
return True
for l10n in self.l10n:
if db.session.is_modified(l10n):
return True
return False
@property
def can_delete(self) -> bool:
"""Returns whether the currency can be deleted.
:return: True if the currency can be deleted, or False otherwise.
"""
from accounting.template_globals import default_currency_code
if self.code == default_currency_code():
return False
return len(self.line_items) == 0
def delete(self) -> None:
"""Deletes the currency.
:return: None.
"""
CurrencyL10n.query.filter(CurrencyL10n.currency == self).delete()
cls: t.Type[t.Self] = self.__class__
cls.query.filter(cls.code == self.code).delete()
class CurrencyL10n(db.Model):
"""A localized currency name."""
__tablename__ = "accounting_currencies_l10n"
"""The table name."""
currency_code = db.Column(db.String,
db.ForeignKey(Currency.code, onupdate="CASCADE",
ondelete="CASCADE"),
nullable=False, primary_key=True)
"""The currency code."""
currency = db.relationship(Currency, back_populates="l10n")
"""The currency."""
locale = db.Column(db.String, nullable=False, primary_key=True)
"""The locale."""
name = db.Column(db.String, nullable=False)
"""The localized name."""
class JournalEntryCurrency:
"""A currency in a journal entry."""
def __init__(self, code: str, debit: list[JournalEntryLineItem],
credit: list[JournalEntryLineItem]):
"""Constructs the currency in the journal entry.
:param code: The currency code.
:param debit: The debit line items.
:param credit: The credit line items.
"""
self.code: str = code
"""The currency code."""
self.debit: list[JournalEntryLineItem] = debit
"""The debit line items."""
self.credit: list[JournalEntryLineItem] = credit
"""The credit line items."""
@property
def name(self) -> str:
"""Returns the currency name.
:return: The currency name.
"""
return db.session.get(Currency, self.code).name
@property
def debit_total(self) -> Decimal:
"""Returns the total amount of the debit line items.
:return: The total amount of the debit line items.
"""
return sum([x.amount for x in self.debit])
@property
def credit_total(self) -> str:
"""Returns the total amount of the credit line items.
:return: The total amount of the credit line items.
"""
return sum([x.amount for x in self.credit])
class JournalEntry(db.Model):
"""A journal entry."""
__tablename__ = "accounting_journal_entries"
"""The table name."""
id = db.Column(db.Integer, nullable=False, primary_key=True,
autoincrement=False)
"""The journal entry ID."""
date = db.Column(db.Date, nullable=False)
"""The date."""
no = db.Column(db.Integer, nullable=False, default=text("1"))
"""The account number under the date."""
note = db.Column(db.String)
"""The note."""
created_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of creation."""
created_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the creator."""
created_by = db.relationship(user_cls, foreign_keys=created_by_id)
"""The creator."""
updated_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of last update."""
updated_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the updator."""
updated_by = db.relationship(user_cls, foreign_keys=updated_by_id)
"""The updator."""
line_items = db.relationship("JournalEntryLineItem",
back_populates="journal_entry")
"""The line items."""
def __str__(self) -> str:
"""Returns the string representation of this journal entry.
:return: The string representation of this journal entry.
"""
if self.is_cash_disbursement:
return gettext("Cash Disbursement Journal Entry#%(id)s",
id=self.id)
if self.is_cash_receipt:
return gettext("Cash Receipt Journal Entry#%(id)s", id=self.id)
return gettext("Transfer Journal Entry#%(id)s", id=self.id)
@property
def currencies(self) -> list[JournalEntryCurrency]:
"""Returns the line items categorized by their currencies.
:return: The currency categories.
"""
line_items: list[JournalEntryLineItem] = sorted(self.line_items,
key=lambda x: x.no)
codes: list[str] = []
by_currency: dict[str, list[JournalEntryLineItem]] = {}
for line_item in line_items:
if line_item.currency_code not in by_currency:
codes.append(line_item.currency_code)
by_currency[line_item.currency_code] = []
by_currency[line_item.currency_code].append(line_item)
return [JournalEntryCurrency(code=x,
debit=[y for y in by_currency[x]
if y.is_debit],
credit=[y for y in by_currency[x]
if not y.is_debit])
for x in codes]
@property
def is_cash_receipt(self) -> bool:
"""Returns whether this is a cash receipt journal entry.
:return: True if this is a cash receipt journal entry, or False
otherwise.
"""
for currency in self.currencies:
if len(currency.debit) > 1:
return False
if currency.debit[0].account.code != Account.CASH_CODE:
return False
return True
@property
def is_cash_disbursement(self) -> bool:
"""Returns whether this is a cash disbursement journal entry.
:return: True if this is a cash disbursement journal entry, or False
otherwise.
"""
for currency in self.currencies:
if len(currency.credit) > 1:
return False
if currency.credit[0].account.code != Account.CASH_CODE:
return False
return True
@property
def can_delete(self) -> bool:
"""Returns whether the journal entry can be deleted.
:return: True if the journal entry can be deleted, or False otherwise.
"""
for line_item in self.line_items:
if len(line_item.offsets) > 0:
return False
return True
def delete(self) -> None:
"""Deletes the journal entry.
:return: None.
"""
JournalEntryLineItem.query\
.filter(JournalEntryLineItem.journal_entry_id == self.id).delete()
db.session.delete(self)
class JournalEntryLineItem(db.Model):
"""A line item in the journal entry."""
__tablename__ = "accounting_journal_entry_line_items"
"""The table name."""
id = db.Column(db.Integer, nullable=False, primary_key=True,
autoincrement=False)
"""The line item ID."""
journal_entry_id = db.Column(db.Integer,
db.ForeignKey(JournalEntry.id,
onupdate="CASCADE",
ondelete="CASCADE"),
nullable=False)
"""The journal entry ID."""
journal_entry = db.relationship(JournalEntry, back_populates="line_items")
"""The journal entry."""
is_debit = db.Column(db.Boolean, nullable=False)
"""True for a debit line item, or False for a credit line item."""
no = db.Column(db.Integer, nullable=False)
"""The line item number under the journal entry and debit or credit."""
original_line_item_id = db.Column(db.Integer,
db.ForeignKey(id, onupdate="CASCADE"),
nullable=True)
"""The ID of the original line item."""
original_line_item = db.relationship("JournalEntryLineItem",
back_populates="offsets",
remote_side=id, passive_deletes=True)
"""The original line item."""
offsets = db.relationship("JournalEntryLineItem",
back_populates="original_line_item")
"""The offset items."""
currency_code = db.Column(db.String,
db.ForeignKey(Currency.code, onupdate="CASCADE"),
nullable=False)
"""The currency code."""
currency = db.relationship(Currency, back_populates="line_items")
"""The currency."""
account_id = db.Column(db.Integer,
db.ForeignKey(Account.id,
onupdate="CASCADE"),
nullable=False)
"""The account ID."""
account = db.relationship(Account, back_populates="line_items", lazy=False)
"""The account."""
description = db.Column(db.String, nullable=True)
"""The description."""
amount = db.Column(db.Numeric(14, 2), nullable=False)
"""The amount."""
def __str__(self) -> str:
"""Returns the string representation of the line item.
:return: The string representation of the line item.
"""
if not hasattr(self, "__str"):
from accounting.template_filters import format_date, format_amount
setattr(self, "__str",
gettext("%(date)s %(description)s %(amount)s",
date=format_date(self.journal_entry.date),
description="" if self.description is None
else self.description,
amount=format_amount(self.amount)))
return getattr(self, "__str")
@property
def account_code(self) -> str:
"""Returns the account code.
:return: The account code.
"""
return self.account.code
@property
def debit(self) -> Decimal | None:
"""Returns the debit amount.
:return: The debit amount, or None if this is not a debit line item.
"""
return self.amount if self.is_debit else None
@property
def is_need_offset(self) -> bool:
"""Returns whether the line item needs offset.
:return: True if the line item needs offset, or False otherwise.
"""
if not self.account.is_need_offset:
return False
if self.account.base_code[0] == "1" and not self.is_debit:
return False
if self.account.base_code[0] == "2" and self.is_debit:
return False
return True
@property
def credit(self) -> Decimal | None:
"""Returns the credit amount.
:return: The credit amount, or None if this is not a credit line item.
"""
return None if self.is_debit else self.amount
@property
def net_balance(self) -> Decimal:
"""Returns the net balance.
:return: The net balance.
"""
if not hasattr(self, "__net_balance"):
setattr(self, "__net_balance", self.amount + sum(
[x.amount if x.is_debit == self.is_debit else -x.amount
for x in self.offsets]))
return getattr(self, "__net_balance")
@net_balance.setter
def net_balance(self, net_balance: Decimal) -> None:
"""Sets the net balance.
:param net_balance: The net balance.
:return: None.
"""
setattr(self, "__net_balance", net_balance)
@property
def query_values(self) -> list[str]:
"""Returns the values to be queried.
:return: The values to be queried.
"""
def format_amount(value: Decimal) -> str:
whole: int = int(value)
frac: Decimal = (value - whole).normalize()
return str(whole) + str(abs(frac))[1:]
return ["{}/{}/{}".format(self.journal_entry.date.year,
self.journal_entry.date.month,
self.journal_entry.date.day),
"" if self.description is None else self.description,
format_amount(self.amount)]
class Option(db.Model):
"""An option."""
__tablename__ = "accounting_options"
"""The table name."""
name = db.Column(db.String, nullable=False, primary_key=True)
"""The name."""
value = db.Column(db.Text, nullable=False)
"""The option value."""
created_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of creation."""
created_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the creator."""
created_by = db.relationship(user_cls, foreign_keys=created_by_id)
"""The creator."""
updated_at = db.Column(db.DateTime(timezone=True), nullable=False,
server_default=db.func.now())
"""The time of last update."""
updated_by_id = db.Column(db.Integer,
db.ForeignKey(user_pk_column,
onupdate="CASCADE"),
nullable=False)
"""The ID of the updator."""
updated_by = db.relationship(user_cls, foreign_keys=updated_by_id)
"""The updator."""