264 lines
9.0 KiB
Python
264 lines
9.0 KiB
Python
import uuid
|
|
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
import brewman.schemas.account as schemas
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.account import Account
|
|
from ..models.account_base import AccountBase
|
|
from ..models.account_type import AccountType
|
|
from ..models.cost_centre import CostCentre
|
|
from ..models.journal import Journal
|
|
from ..models.voucher import Voucher
|
|
from ..models.voucher_type import VoucherType
|
|
from ..schemas.balance import AccountBalance
|
|
from ..schemas.user import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_model=None)
|
|
def save(
|
|
data: schemas.AccountIn,
|
|
user: UserToken = Security(get_user, scopes=["accounts"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item = Account(
|
|
name=data.name,
|
|
code=Account.get_code(data.type_, db),
|
|
type_id=data.type_,
|
|
is_starred=data.is_starred,
|
|
is_active=data.is_active,
|
|
is_reconcilable=data.is_reconcilable,
|
|
cost_centre_id=data.cost_centre.id_,
|
|
)
|
|
|
|
db.add(item)
|
|
db.commit()
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.put("/{id_}", response_model=None)
|
|
def update_route(
|
|
id_: uuid.UUID,
|
|
data: schemas.AccountIn,
|
|
user: UserToken = Security(get_user, scopes=["accounts"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Account = db.execute(select(Account).where(Account.id == id_)).scalar_one()
|
|
if item.is_fixture:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_423_LOCKED,
|
|
detail=f"{item.name} is a fixture and cannot be edited or deleted.",
|
|
)
|
|
if not item.type_id == data.type_:
|
|
item.code = Account.get_code(data.type_, db)
|
|
item.type_id = data.type_
|
|
item.name = data.name
|
|
item.is_active = data.is_active
|
|
item.is_reconcilable = data.is_reconcilable
|
|
item.is_starred = data.is_starred
|
|
item.cost_centre_id = data.cost_centre.id_
|
|
db.commit()
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.delete("/{id_}", response_model=schemas.AccountBlank)
|
|
def delete_route(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["accounts"]),
|
|
) -> schemas.AccountBlank:
|
|
with SessionFuture() as db:
|
|
account: Account = db.execute(select(Account).where(Account.id == id_)).scalar_one()
|
|
can_delete, reason = account.can_delete("advanced-delete" in user.permissions)
|
|
if can_delete:
|
|
delete_with_data(account, db)
|
|
db.commit()
|
|
return account_blank(db)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Cannot delete account because {reason}",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=schemas.AccountBlank)
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["accounts"]),
|
|
) -> schemas.AccountBlank:
|
|
with SessionFuture() as db:
|
|
return account_blank(db)
|
|
|
|
|
|
@router.get("/list", response_model=list[schemas.Account])
|
|
async def show_list(user: UserToken = Depends(get_user)) -> list[schemas.Account]:
|
|
with SessionFuture() as db:
|
|
return [
|
|
account_info(item)
|
|
for item in db.execute(
|
|
select(Account).order_by(Account.type_id).order_by(Account.name).order_by(Account.code)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
]
|
|
|
|
|
|
@router.get("/query", response_model=list[schemas.AccountLink])
|
|
async def show_term(
|
|
q: str,
|
|
t: int | None = None, # AccountType
|
|
r: bool | None = None, # Reconcilable
|
|
a: bool | None = None, # Active
|
|
c: int | None = None, # Count
|
|
current_user: UserToken = Depends(get_user),
|
|
) -> list[schemas.AccountLink]:
|
|
list_: list[schemas.AccountLink] = []
|
|
with SessionFuture() as db:
|
|
query_ = select(AccountBase)
|
|
if t is not None:
|
|
query_ = query_.where(AccountBase.type_id == t)
|
|
if r is not None:
|
|
query_ = query_.where(AccountBase.is_reconcilable == r)
|
|
if a is not None:
|
|
query_ = query_.where(AccountBase.is_active == a)
|
|
if q is not None:
|
|
for name in q.split():
|
|
query_ = query_.where(AccountBase.name.ilike(f"%{name}%"))
|
|
query_ = query_.order_by(AccountBase.name)
|
|
if c is not None:
|
|
query_ = query_.limit(c)
|
|
data: list[AccountBase] = list(db.execute(query_).scalars().all())
|
|
|
|
for item in data:
|
|
list_.append(schemas.AccountLink(id_=item.id, name=item.name))
|
|
return list_
|
|
|
|
|
|
@router.get("/{id_}/balance", response_model=AccountBalance)
|
|
async def show_balance(
|
|
id_: uuid.UUID,
|
|
d: str | None = None,
|
|
user: UserToken = Depends(get_user),
|
|
) -> AccountBalance:
|
|
date_ = None if d is None or d == "" else datetime.strptime(d, "%d-%b-%Y")
|
|
with SessionFuture() as db:
|
|
return AccountBalance(date_=balance(id_, date_, db), total=balance(id_, None, db))
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.Account)
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["accounts"]),
|
|
) -> schemas.Account:
|
|
with SessionFuture() as db:
|
|
item: Account = db.execute(select(Account).where(Account.id == id_)).scalar_one()
|
|
return account_info(item)
|
|
|
|
|
|
def balance(id_: uuid.UUID, date_: date | None, db: Session) -> Decimal:
|
|
account: AccountBase = db.execute(select(AccountBase).where(AccountBase.id == id_)).scalar_one()
|
|
if not account.type_.balance_sheet:
|
|
return Decimal(0)
|
|
|
|
bal = select(func.sum(Journal.amount * Journal.debit)).join(Journal.voucher)
|
|
if date_ is not None:
|
|
bal = bal.where(Voucher.date_ <= date_)
|
|
|
|
bal = bal.where(Voucher.voucher_type != VoucherType.ISSUE).where(Journal.account_id == id_)
|
|
result: Decimal | None = db.execute(bal).scalar_one_or_none()
|
|
return Decimal(0) if result is None else result
|
|
|
|
|
|
def account_info(item: Account) -> schemas.Account:
|
|
return schemas.Account(
|
|
id_=item.id,
|
|
code=item.code,
|
|
name=item.name,
|
|
type_=item.type_id,
|
|
is_active=item.is_active,
|
|
is_reconcilable=item.is_reconcilable,
|
|
is_starred=item.is_starred,
|
|
is_fixture=item.is_fixture,
|
|
cost_centre=schemas.CostCentreLink(
|
|
id_=item.cost_centre_id,
|
|
name=item.cost_centre.name,
|
|
),
|
|
)
|
|
|
|
|
|
def account_blank(db: Session) -> schemas.AccountBlank:
|
|
return schemas.AccountBlank(
|
|
name="",
|
|
type_=db.execute(select(AccountType.id).where(AccountType.name == "Creditors")).scalar_one(),
|
|
is_active=True,
|
|
is_reconcilable=False,
|
|
is_starred=False,
|
|
cost_centre=CostCentre.overall(),
|
|
is_fixture=False,
|
|
)
|
|
|
|
|
|
def delete_with_data(account: Account, db: Session) -> None:
|
|
suspense_account = db.execute(select(Account).where(Account.id == Account.suspense())).scalar_one()
|
|
query: list[Voucher] = list(
|
|
db.execute(
|
|
select(Voucher)
|
|
.options(joinedload(Voucher.journals, innerjoin=True).joinedload(Journal.account, innerjoin=True))
|
|
.where(Voucher.journals.any(Journal.account_id == account.id))
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
for voucher in query:
|
|
others, sus_jnl, acc_jnl = False, None, None
|
|
for journal in voucher.journals:
|
|
if journal.account_id == account.id:
|
|
acc_jnl = journal
|
|
elif journal.account_id == Account.suspense():
|
|
sus_jnl = journal
|
|
else:
|
|
others = True
|
|
if not others:
|
|
db.delete(voucher)
|
|
else:
|
|
assert acc_jnl is not None
|
|
if sus_jnl is None:
|
|
acc_jnl.account = suspense_account
|
|
voucher.narration += f"\nSuspense \u20B9{acc_jnl.amount:,.2f} is {account.name}"
|
|
else:
|
|
amount = (sus_jnl.debit * sus_jnl.amount) + (acc_jnl.debit * acc_jnl.amount)
|
|
db.delete(acc_jnl)
|
|
if amount == 0:
|
|
db.delete(sus_jnl)
|
|
else:
|
|
sus_jnl.amount = abs(amount)
|
|
sus_jnl.debit = -1 if amount < 0 else 1
|
|
voucher.narration += f"\nDeleted \u20B9{acc_jnl.amount * acc_jnl.debit:,.2f} of {account.name}"
|
|
if voucher.voucher_type in (
|
|
VoucherType.PAYMENT,
|
|
VoucherType.RECEIPT,
|
|
):
|
|
voucher.voucher_type = VoucherType.JOURNAL
|
|
db.delete(account)
|