brewman/brewman/brewman/routers/reports/cash_flow.py

from datetime import date, datetime
from typing import List, Tuple

import brewman.schemas.cash_flow as schemas

from fastapi import APIRouter, Request, Security
from sqlalchemy import not_
from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import desc, func, select

from ...core.security import get_current_active_user as get_user
from ...core.session import get_finish_date, get_start_date, set_period
from ...db.session import SessionFuture
from ...models.account_base import AccountBase
from ...models.account_type import AccountType
from ...models.journal import Journal
from ...models.voucher import Voucher
from ...models.voucher_type import VoucherType
from ...schemas.user import UserToken


router = APIRouter()


@router.get("", response_model=schemas.CashFlow)
def report_blank(
    request: Request,
    user: UserToken = Security(get_user, scopes=["cash-flow"]),
) -> schemas.CashFlow:
    return schemas.CashFlow(
        startDate=get_start_date(request.session),
        finishDate=get_finish_date(request.session),
        body=schemas.CashFlowBody(operating=[], investing=[], financing=[], details=[]),
        footer=[],
    )


@router.get("/data", response_model=schemas.CashFlow)
def report_data(
    request: Request,
    s: str,
    f: str,
    user: UserToken = Security(get_user, scopes=["cash-flow"]),
) -> schemas.CashFlow:
    with SessionFuture() as db:
        start_date = datetime.strptime(s, "%d-%b-%Y")
        finish_date = datetime.strptime(f, "%d-%b-%Y")
        body, footer = build_report(start_date, finish_date, db)
        set_period(s, f, request.session)
        return schemas.CashFlow(
            startDate=s,
            finishDate=f,
            body=body,
            footer=footer,
        )


@router.get("/{id_}", response_model=schemas.CashFlow)
def report_id(
    id_: int,
    request: Request,
    s: str,
    f: str,
    user: UserToken = Security(get_user, scopes=["cash-flow"]),
) -> schemas.CashFlow:
    with SessionFuture() as db:
        start_date = datetime.strptime(s, "%d-%b-%Y")
        finish_date = datetime.strptime(f, "%d-%b-%Y")
        body, footer = build_report_id(id_, start_date, finish_date, db)
        set_period(s, f, request.session)
        return schemas.CashFlow(
            startDate=s,
            finishDate=f,
            body=body,
            footer=footer,
        )


def build_report(
    start_date: date, finish_date: date, db: Session
) -> Tuple[schemas.CashFlowBody, List[schemas.CashFlowItem]]:
    sub_query = (
        select(Voucher.id)
        .join(Voucher.journals)
        .join(Journal.account)
        .where(
            AccountBase.type_id == select(AccountType.id).where(AccountType.name == "Cash").scalar_subquery(),
            Voucher.date >= start_date,
            Voucher.date <= finish_date,
        )
    )

    query = db.execute(
        select(AccountType, func.sum(Journal.signed_amount))
        .join(Journal.voucher)
        .join(Journal.account)
        .join(AccountBase.type_)
        .where(
            Voucher.id.in_(sub_query),
            AccountBase.type_id != select(AccountType.id).where(AccountType.name == "Cash").scalar_subquery(),
        )
        .group_by(AccountType)
        .order_by(func.sum(Journal.signed_amount))
    ).all()

    total_amount = 0
    cf: schemas.CashFlowBody = schemas.CashFlowBody(operating=[], investing=[], financing=[], details=[])
    dict_ = {"operating": cf.operating, "investing": cf.investing, "financing": cf.investing, "details": cf.details}
    for account_type, amount in query:
        total_amount += amount * -1
        dict_[account_type.cash_flow_classification.lower()].append(
            schemas.CashFlowItem(
                name=account_type.name,
                url=["/", "cash-flow", str(account_type.id)],
                amount=amount * -1,
            )
        )

    opening = db.execute(
        select(func.sum(Journal.amount * Journal.debit))
        .join(Journal.voucher)
        .join(Journal.account)
        .where(
            Voucher.date < start_date,
            not_(Voucher.voucher_type.in_([VoucherType.ISSUE, VoucherType.CLOSING_STOCK])),
            AccountBase.type_id == select(AccountType.id).where(AccountType.name == "Cash"),
        )
    ).scalar()

    closing = db.execute(
        select(func.sum(Journal.amount * Journal.debit))
        .join(Journal.voucher)
        .join(Journal.account)
        .where(
            Voucher.date <= finish_date,
            not_(Voucher.voucher_type.in_([VoucherType.ISSUE, VoucherType.CLOSING_STOCK])),
            AccountBase.type_id == select(AccountType.id).where(AccountType.name == "Cash"),
        )
    ).scalar()

    return (
        cf,
        [
            schemas.CashFlowItem(
                name="Net increase in cash and cash equivalents",
                url=[],
                amount=total_amount,
            ),
            schemas.CashFlowItem(
                name="Cash and cash equivalents at beginning of period",
                url=[],
                amount=opening,
            ),
            schemas.CashFlowItem(name="Cash and cash equivalents at end of period", url=[], amount=closing),
        ],
    )


def build_report_id(
    account_type: int, start_date: date, finish_date: date, db: Session
) -> Tuple[schemas.CashFlowBody, List[schemas.CashFlowItem]]:
    cf: schemas.CashFlowBody = schemas.CashFlowBody(operating=[], investing=[], financing=[], details=[])
    details = cf.details
    sub_query = (
        select(Voucher.id)
        .join(Voucher.journals)
        .join(Journal.account)
        .where(
            AccountBase.type_id == select(AccountType.id).where(AccountType.name == "Cash").scalar_subquery(),
            Voucher.date >= start_date,
            Voucher.date <= finish_date,
        )
    )

    query = db.execute(
        select(AccountBase, func.sum(Journal.signed_amount))
        .join(Journal, Voucher.journals)
        .join(AccountBase, Journal.account)
        .where(Voucher.id.in_(sub_query), AccountBase.type_id == account_type)
        .group_by(AccountBase)
        .order_by(desc(func.sum(Journal.amount)))
    ).all()

    total_amount = 0
    for account, amount in query:
        total_amount += amount * -1
        details.append(
            schemas.CashFlowItem(
                name=account.name,
                url=["/", "ledger", str(account.id)],
                amount=amount * -1,
            )
        )

    return cf, [schemas.CashFlowItem(name="total", url=[], amount=total_amount)]