Added ledger.

This commit is contained in:
2023-03-05 14:25:00 +08:00
parent fe77f87110
commit eabe80b790
12 changed files with 736 additions and 86 deletions

View File

@ -0,0 +1,34 @@
# The Mia! Accounting Flask Project.
# Author: imacat@mail.imacat.idv.tw (imacat), 2023/3/5
# 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 option link.
"""
class OptionLink:
"""An option link."""
def __init__(self, title: str, url: str, is_active: bool):
"""Constructs an option link.
:param title: The title.
:param url: The URI.
:param is_active: True if active, or False otherwise
"""
self.title: str = title
self.url: str = url
self.is_active: bool = is_active

View File

@ -26,7 +26,7 @@ from datetime import date
from flask import url_for
from accounting.models import Transaction
from accounting.models import Currency, Account, Transaction
from .period import YearPeriod, Period, ThisMonth, LastMonth, SinceLastMonth, \
ThisYear, LastYear, Today, Yesterday, TemplatePeriod
@ -116,3 +116,26 @@ class JournalPeriodChooser(PeriodChooser):
if period.is_default:
return url_for("accounting.report.journal-default")
return url_for("accounting.report.journal", period=period)
class LedgerPeriodChooser(PeriodChooser):
"""The ledger period chooser."""
def __init__(self, currency: Currency, account: Account):
"""Constructs the ledger period chooser."""
self.currency: Currency = currency
"""The currency."""
self.account: Account = account
"""The account."""
first: Transaction | None \
= Transaction.query.order_by(Transaction.date).first()
super(LedgerPeriodChooser, self).__init__(
None if first is None else first.date)
def _url_for(self, period: Period) -> str:
if period.is_default:
return url_for("accounting.report.ledger-default",
currency=self.currency, account=self.account)
return url_for("accounting.report.ledger",
currency=self.currency, account=self.account,
period=period)

View File

@ -27,40 +27,26 @@ from flask_babel import LazyString
from accounting import db
from accounting.locale import gettext
from accounting.models import Currency
from accounting.models import Currency, Account
from accounting.template_globals import default_currency_code
from .option_link import OptionLink
from .period import Period
from .report_type import ReportType
class ReportLink:
"""A link of a report."""
def __init__(self, name: str | LazyString, url: str):
"""Constructs a report.
:param name: The report name.
:param url: The URL.
"""
self.name: str | LazyString = name
"""The report name."""
self.url: str = url
"""The URL."""
self.is_active: bool = False
"""Whether the report is the current report."""
class ReportChooser:
"""The report chooser."""
def __init__(self, active_report: ReportType,
period: Period | None = None,
currency: Currency | None = None):
currency: Currency | None = None,
account: Account | None = None):
"""Constructs the report chooser.
:param active_report: The active report.
:param period: The period.
:param currency: The currency.
:param account: The account.
"""
self.__active_report: ReportType = active_report
"""The currently active report."""
@ -71,17 +57,21 @@ class ReportChooser:
Currency, default_currency_code()) \
if currency is None else currency
"""The currency."""
self.__reports: list[ReportLink] = []
self.__account: Account = Account.find_by_code("1111-001") \
if account is None else account
"""The currency."""
self.__reports: list[OptionLink] = []
"""The links to the reports."""
self.__reports.append(self.__journal)
self.current_report: str | LazyString = ""
"""The name of the current report."""
"""The title of the current report."""
self.__reports.append(self.__journal)
self.__reports.append(self.__ledger)
for report in self.__reports:
if report.is_active:
self.current_report = report.name
self.current_report = report.title
@property
def __journal(self) -> ReportLink:
def __journal(self) -> OptionLink:
"""Returns the journal.
:return: The journal.
@ -89,12 +79,25 @@ class ReportChooser:
url: str = url_for("accounting.report.journal-default") \
if self.__period.is_default \
else url_for("accounting.report.journal", period=self.__period)
report = ReportLink(gettext("Journal"), url)
if self.__active_report == ReportType.JOURNAL:
report.is_active = True
return report
return OptionLink(gettext("Journal"), url,
self.__active_report == ReportType.JOURNAL)
def __iter__(self) -> t.Iterator[ReportLink]:
@property
def __ledger(self) -> OptionLink:
"""Returns the ledger.
:return: The ledger.
"""
url: str = url_for("accounting.report.ledger-default",
currency=self.__currency, account=self.__account) \
if self.__period.is_default \
else url_for("accounting.report.ledger",
currency=self.__currency, account=self.__account,
period=self.__period)
return OptionLink(gettext("Ledger"), url,
self.__active_report == ReportType.LEDGER)
def __iter__(self) -> t.Iterator[OptionLink]:
"""Returns the iteration of the reports.
:return: The iteration of the reports.

View File

@ -18,8 +18,9 @@
"""
import typing as t
from decimal import Decimal
from abc import ABC, abstractmethod
from datetime import date
from decimal import Decimal
from accounting.models import JournalEntry, Transaction, Account, Currency
@ -38,27 +39,77 @@ class ReportRow(ABC):
class JournalRow(ReportRow):
"""A row in the journal report."""
def __init__(self, entry: JournalEntry, transaction: Transaction,
account: Account, currency: Currency):
def __init__(self, entry: JournalEntry):
"""Constructs the row in the journal report.
:param entry: The journal entry.
:param transaction: The transaction.
:param account: The account.
:param currency: The currency.
"""
self.is_debit: bool = entry.is_debit
self.entry: JournalEntry = entry
"""The journal entry."""
self.summary: str | None = entry.summary
"""The summary."""
self.currency_code: str = entry.currency_code
"""The currency code."""
self.is_debit: bool = entry.is_debit
"""True for a debit journal entry, or False for a credit entry."""
self.amount: Decimal = entry.amount
self.transaction: Account = transaction
self.account: Account = account
self.currency: Currency = currency
"""The amount."""
self.transaction: Transaction | None = None
"""The transaction."""
self.currency: Currency | None = None
"""The currency."""
self.account: Account | None = None
"""The account."""
def as_dict(self) -> dict[str, t.Any]:
return {"date": self.transaction.date.isoformat(),
"currency": self.currency.name,
return {"date": self.transaction.date,
"currency": str(self.currency),
"account": str(self.account),
"summary": self.summary,
"debit": self.amount if self.is_debit else None,
"credit": None if self.is_debit else self.amount,
"note": self.transaction.note}
class LedgerRow(ReportRow):
"""A row in the ledger report."""
def __init__(self, entry: JournalEntry | None = None):
"""Constructs the row in the journal report.
:param entry: The journal entry.
"""
self.entry: JournalEntry | None = None
"""The journal entry."""
self.transaction: Transaction | None = None
"""The transaction."""
self.is_total: bool = False
"""Whether this is the total row."""
self.date: date | None = None
"""The date."""
self.summary: str | None = None
"""The summary."""
self.debit: Decimal | None = None
"""The debit amount."""
self.credit: Decimal | None = None
"""The credit amount."""
self.balance: Decimal | None = None
"""The balance."""
if entry is not None:
self.entry = entry
self.summary = entry.summary
self.debit = entry.amount if entry.is_debit else None
self.credit = None if entry.is_debit else entry.amount
def as_dict(self) -> dict[str, t.Any]:
if self.is_total:
return {"date": "Total",
"summary": None,
"debit": self.debit,
"credit": self.credit,
"balance": self.balance}
return {"date": self.date,
"summary": self.summary,
"debit": self.debit,
"credit": self.credit,
"balance": self.balance}

View File

@ -24,3 +24,5 @@ class ReportType(Enum):
"""The report types."""
JOURNAL: str = "journal"
"""The journal."""
LEDGER: str = "ledger"
"""The ledger."""

View File

@ -20,24 +20,32 @@
import csv
import typing as t
from abc import ABC, abstractmethod
from decimal import Decimal
from io import StringIO
from urllib.parse import urlparse, ParseResult, parse_qsl, urlencode, \
urlunparse
import sqlalchemy as sa
from flask import Response, render_template
from flask_sqlalchemy.query import Query
from flask import Response, render_template, request, url_for
from accounting import db
from accounting.models import JournalEntry, Transaction, Account, Currency
from accounting.locale import gettext
from accounting.models import Currency, Account, Transaction, JournalEntry
from accounting.utils.pagination import Pagination
from accounting.utils.txn_types import TransactionType
from .option_link import OptionLink
from .period import Period
from .period_choosers import PeriodChooser, JournalPeriodChooser
from .period_choosers import PeriodChooser, JournalPeriodChooser, \
LedgerPeriodChooser
from .report_chooser import ReportChooser
from .report_rows import ReportRow, JournalRow
from .report_rows import ReportRow, JournalRow, LedgerRow
from .report_type import ReportType
T = t.TypeVar("T", bound=ReportRow)
"""The row class in the report."""
class JournalEntryReport(ABC):
class JournalEntryReport(t.Generic[T], ABC):
"""A report based on a journal entry."""
def __init__(self, period: Period):
@ -47,22 +55,23 @@ class JournalEntryReport(ABC):
"""
self.period: Period = period
"""The period."""
self._entries: list[JournalEntry] = self.get_entries()
"""The journal entries."""
self.__rows: list[T] | None = None
"""The rows in the report."""
@abstractmethod
def get_entries(self) -> list[JournalEntry]:
"""Returns the journal entries.
def get_rows(self) -> list[T]:
"""Returns the rows, without pagination.
:return: The journal entries.
:return: The rows.
"""
@abstractmethod
def entries_to_rows(self, entries: list[JournalEntry]) -> list[ReportRow]:
"""Converts the journal entries into report rows.
def populate_rows(self, rows: list[T]) -> None:
"""Populates the transaction, currency, account, and other data to the
given rows.
:param entries: The journal entries.
:return: The report rows.
:param rows: The rows.
:return: None.
"""
@property
@ -104,6 +113,16 @@ class JournalEntryReport(ABC):
:return: The report as an HTML page.
"""
@property
def rows(self) -> list[T]:
"""Returns the journal entries.
:return: The journal entries.
"""
if self.__rows is None:
self.__rows = self.get_rows()
return self.__rows
@property
def txn_types(self) -> t.Type[TransactionType]:
"""Returns the transaction types.
@ -112,17 +131,32 @@ class JournalEntryReport(ABC):
"""
return TransactionType
@property
def csv_uri(self) -> str:
"""Returns the URI to download the report as CSV.
:return: The URI to download the report as CSV.
"""
uri: str = request.full_path if request.query_string else request.path
uri_p: ParseResult = urlparse(uri)
params: list[tuple[str, str]] = parse_qsl(uri_p.query)
params = [x for x in params if x[0] != "as"]
params.append(("as", "csv"))
parts: list[str] = list(uri_p)
parts[4] = urlencode(params)
return urlunparse(parts)
def as_csv_download(self) -> Response:
"""Returns the journal entries as CSV download.
"""Returns the report as CSV download.
:return: The CSV download response.
"""
self.populate_rows(self.rows)
with StringIO() as fp:
writer: csv.DictWriter = csv.DictWriter(
fp, fieldnames=self.csv_field_names)
writer.writeheader()
writer.writerows([x.as_dict()
for x in self.entries_to_rows(self._entries)])
writer.writerows([x.as_dict() for x in self.rows])
fp.seek(0)
response: Response = Response(fp.read(), mimetype="text/csv")
response.headers["Content-Disposition"] \
@ -130,35 +164,37 @@ class JournalEntryReport(ABC):
return response
class Journal(JournalEntryReport):
class Journal(JournalEntryReport[JournalRow]):
"""The journal."""
def get_entries(self) -> list[JournalEntry]:
def get_rows(self) -> list[JournalRow]:
conditions: list[sa.BinaryExpression] = []
if self.period.start is not None:
conditions.append(Transaction.date >= self.period.start)
if self.period.end is not None:
conditions.append(Transaction.date <= self.period.end)
query: Query = db.session.query(JournalEntry).join(Transaction)
if len(conditions) > 0:
query = query.filter(*conditions)
return query.order_by(Transaction.date,
JournalEntry.is_debit.desc(),
JournalEntry.no).all()
return [JournalRow(x) for x in db.session
.query(JournalEntry)
.join(Transaction)
.filter(*conditions)
.order_by(Transaction.date,
JournalEntry.is_debit.desc(),
JournalEntry.no).all()]
def entries_to_rows(self, entries: list[JournalEntry]) -> list[ReportRow]:
def populate_rows(self, rows: list[JournalRow]) -> None:
transactions: dict[int, Transaction] \
= {x.id: x for x in Transaction.query.filter(
Transaction.id.in_({x.transaction_id for x in entries}))}
Transaction.id.in_({x.entry.transaction_id for x in rows}))}
accounts: dict[int, Account] \
= {x.id: x for x in Account.query.filter(
Account.id.in_({x.account_id for x in entries}))}
Account.id.in_({x.entry.account_id for x in rows}))}
currencies: dict[int, Currency] \
= {x.code: x for x in Currency.query.filter(
Currency.code.in_({x.currency_code for x in entries}))}
return [JournalRow(x, transactions[x.transaction_id],
accounts[x.account_id], currencies[x.currency_code])
for x in entries]
Currency.code.in_({x.entry.currency_code for x in rows}))}
for row in rows:
row.transaction = transactions[row.entry.transaction_id]
row.account = accounts[row.entry.account_id]
row.currency = currencies[row.entry.currency_code]
@property
def csv_field_names(self) -> list[str]:
@ -175,10 +211,208 @@ class Journal(JournalEntryReport):
@property
def report_chooser(self) -> ReportChooser:
return ReportChooser(ReportType.JOURNAL, self.period)
return ReportChooser(ReportType.JOURNAL,
period=self.period)
def as_html_page(self) -> str:
pagination: Pagination = Pagination[JournalEntry](self._entries)
pagination: Pagination = Pagination[JournalRow](self.rows)
rows: list[JournalRow] = pagination.list
self.populate_rows(rows)
return render_template("accounting/report/journal.html",
list=self.entries_to_rows(pagination.list),
pagination=pagination, report=self)
list=rows, pagination=pagination, report=self)
class Ledger(JournalEntryReport[LedgerRow]):
"""The ledger."""
def __init__(self, currency: Currency, account: Account, period: Period):
"""Constructs a journal.
:param currency: The currency.
:param account: The account.
:param period: The period.
"""
super().__init__(period)
self.currency: Currency = currency
"""The currency."""
self.account: Account = account
"""The account."""
self.total_row: LedgerRow | None = None
"""The total row to show on the template."""
def get_rows(self) -> list[LedgerRow]:
brought_forward: LedgerRow | None = self.__get_brought_forward_row()
rows: list[LedgerRow] = [LedgerRow(x) for x in self.__query_entries()]
total: LedgerRow = self.__get_total_row(brought_forward, rows)
self.__populate_balance(brought_forward, rows)
if brought_forward is not None:
rows.insert(0, brought_forward)
rows.append(total)
return rows
def __get_brought_forward_row(self) -> LedgerRow | None:
"""Queries, composes and returns the brought-forward row.
:return: The brought-forward row, or None if the ledger starts from the
beginning.
"""
if self.period.start is None:
return None
balance_func: sa.Function = sa.func.sum(sa.case(
(JournalEntry.is_debit, JournalEntry.amount),
else_=-JournalEntry.amount))
select: sa.Select = sa.Select(balance_func).join(Transaction)\
.filter(JournalEntry.currency_code == self.currency.code,
JournalEntry.account_id == self.account.id,
Transaction.date < self.period.start)
balance: int | None = db.session.scalar(select)
if balance is None:
return None
row: LedgerRow = LedgerRow()
row.date = self.period.start
row.summary = gettext("Brought forward")
if balance > 0:
row.debit = balance
elif balance < 0:
row.credit = -balance
row.balance = balance
return row
def __query_entries(self) -> list[JournalEntry]:
"""Queries and returns the journal entries.
:return: The journal entries.
"""
conditions: list[sa.BinaryExpression] \
= [JournalEntry.currency_code == self.currency.code,
JournalEntry.account_id == self.account.id]
if self.period.start is not None:
conditions.append(Transaction.date >= self.period.start)
if self.period.end is not None:
conditions.append(Transaction.date <= self.period.end)
return db.session.query(JournalEntry).join(Transaction)\
.filter(*conditions)\
.order_by(Transaction.date,
JournalEntry.is_debit.desc(),
JournalEntry.no).all()
@staticmethod
def __get_total_row(brought_forward: LedgerRow | None,
rows: list[LedgerRow]) -> LedgerRow:
"""Composes the total row.
:param brought_forward: The brought-forward row.
:param rows: The rows.
:return: None.
"""
row: LedgerRow = LedgerRow()
row.is_total = True
row.summary = gettext("Total")
row.debit = sum([x.debit for x in rows if x.debit is not None])
row.credit = sum([x.credit for x in rows if x.credit is not None])
row.balance = row.debit - row.credit
if brought_forward is not None:
row.balance = brought_forward.balance + row.balance
return row
@staticmethod
def __populate_balance(brought_forward: LedgerRow | None,
rows: list[LedgerRow]) -> None:
"""Populates the balance of the rows.
:param brought_forward: The brought-forward row.
:param rows: The rows.
:return: None.
"""
balance: Decimal = 0 if brought_forward is None \
else brought_forward.balance
for row in rows:
if row.debit is not None:
balance = balance + row.debit
if row.credit is not None:
balance = balance - row.credit
row.balance = balance
def populate_rows(self, rows: list[LedgerRow]) -> None:
transactions: dict[int, Transaction] \
= {x.id: x for x in Transaction.query.filter(
Transaction.id.in_({x.entry.transaction_id for x in rows
if x.entry is not None}))}
for row in rows:
if row.entry is not None:
row.transaction = transactions[row.entry.transaction_id]
row.date = row.transaction.date
@property
def csv_field_names(self) -> list[str]:
return ["date", "summary", "debit", "credit", "balance"]
@property
def csv_filename(self) -> str:
return "ledger-{currency}-{account}-{period}.csv".format(
currency=self.currency.code, account=self.account.code,
period=self.period.spec)
@property
def period_chooser(self) -> PeriodChooser:
return LedgerPeriodChooser(self.currency, self.account)
@property
def report_chooser(self) -> ReportChooser:
return ReportChooser(ReportType.LEDGER,
currency=self.currency,
account=self.account,
period=self.period)
def as_html_page(self) -> str:
pagination: Pagination = Pagination[LedgerRow](self.rows)
rows: list[LedgerRow] = pagination.list
self.populate_rows(rows)
if len(rows) > 0 and rows[-1].is_total:
self.total_row = rows[-1]
rows = rows[:-1]
return render_template("accounting/report/ledger.html",
list=rows, pagination=pagination, report=self)
@property
def currency_options(self) -> list[OptionLink]:
"""Returns the currency options.
:return: The currency options.
"""
def get_url(currency: Currency):
if self.period.is_default:
return url_for("accounting.report.ledger-default",
currency=currency, account=self.account)
return url_for("accounting.report.ledger",
currency=currency, account=self.account,
period=self.period)
in_use: set[str] = set(db.session.scalars(
sa.select(JournalEntry.currency_code)
.group_by(JournalEntry.currency_code)).all())
return [OptionLink(str(x), get_url(x), x.code == self.currency.code)
for x in Currency.query.filter(Currency.code.in_(in_use))
.order_by(Currency.code).all()]
@property
def account_options(self) -> list[OptionLink]:
"""Returns the account options.
:return: The account options.
"""
def get_url(account: Account):
if self.period.is_default:
return url_for("accounting.report.ledger-default",
currency=self.currency, account=account)
return url_for("accounting.report.ledger",
currency=self.currency, account=account,
period=self.period)
in_use: set[int] = set(db.session.scalars(
sa.select(JournalEntry.account_id)
.filter(JournalEntry.currency_code == self.currency.code)
.group_by(JournalEntry.account_id)).all())
return [OptionLink(str(x), get_url(x), x.id == self.account.id)
for x in Account.query.filter(Account.id.in_(in_use))
.order_by(Account.base_code, Account.no).all()]

View File

@ -19,9 +19,10 @@
"""
from flask import Blueprint, request, Response
from accounting.models import Currency, Account
from accounting.utils.permission import has_permission, can_view
from .period import Period
from .reports import Journal
from .reports import Journal, Ledger
bp: Blueprint = Blueprint("report", __name__)
"""The view blueprint for the reports."""
@ -58,3 +59,47 @@ def __get_journal_list(period: Period) -> str | Response:
if "as" in request.args and request.args["as"] == "csv":
return report.as_csv_download()
return report.as_html_page()
@bp.get("ledger/<currency:currency>/<account:account>",
endpoint="ledger-default")
@has_permission(can_view)
def get_default_ledger_list(currency: Currency, account: Account) \
-> str | Response:
"""Returns the ledger in the default period.
:param currency: The currency.
:param account: The account.
:return: The ledger in the default period.
"""
return __get_ledger_list(currency, account, Period.get_instance())
@bp.get("ledger/<currency:currency>/<account:account>/<period:period>",
endpoint="ledger")
@has_permission(can_view)
def get_ledger_list(currency: Currency, account: Account, period: Period) \
-> str | Response:
"""Returns the ledger.
:param currency: The currency.
:param account: The account.
:param period: The period.
:return: The ledger in the period.
"""
return __get_ledger_list(currency, account, period)
def __get_ledger_list(currency: Currency, account: Account, period: Period) \
-> str | Response:
"""Returns the ledger.
:param currency: The currency.
:param account: The account.
:param period: The period.
:return: The ledger in the period.
"""
report: Ledger = Ledger(currency, account, period)
if "as" in request.args and request.args["as"] == "csv":
return report.as_csv_download()
return report.as_html_page()