import uuid from typing import List, Optional from datetime import datetime from fastapi import APIRouter, HTTPException, status, Depends, Security from sqlalchemy import func from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import joinedload_all, Session from ..schemas.auth import UserToken from ..core.security import get_current_active_user as get_user from ..db.session import SessionLocal from ..models.master import CostCentre, Account, AccountType, AccountBase from ..models.voucher import Voucher, Journal, VoucherType import brewman.schemas.master as schemas router = APIRouter() # Dependency def get_db() -> Session: try: db = SessionLocal() yield db finally: db.close() @router.post("", response_model=schemas.Account) def save( data: schemas.AccountIn, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["accounts"]), ): try: item = Account( name=data.name, type_=data.type, is_starred=data.is_starred, is_active=data.is_active, is_reconcilable=data.is_reconcilable, cost_centre_id=data.cost_centre.id_, ).create(db) db.commit() return account_info(item) except SQLAlchemyError as e: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e), ) except Exception: db.rollback() raise @router.put("/{id_}", response_model=schemas.Account) def update( id_: uuid.UUID, data: schemas.AccountIn, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["accounts"]), ): try: item: Account = db.query(Account).filter(Account.id == id_).first() 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 == data.type: item.code = Account.get_code(data.type, db) item.type = 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() return account_info(item) except SQLAlchemyError as e: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e), ) except Exception: db.rollback() raise @router.delete("/{id}") def delete( id_: uuid.UUID, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["accounts"]), ): account: Account = db.query(Account).filter(Account.id == id_).first() can_delete, reason = account.can_delete("advanced-delete" in user.permissions) if can_delete: delete_with_data(account, db) db.commit() return account_info(None) else: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Cannot delete account because {reason}", ) @router.get("") def show_blank( db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["accounts"]), ): return account_info(None) @router.get("/list") async def show_list(db: Session = Depends(get_db), user: UserToken = Depends(get_user)): return [ { "id": item.id, "name": item.name, "type": item.type_object.name, "isActive": item.is_active, "isReconcilable": item.is_reconcilable, "isStarred": item.is_starred, "costCentre": item.cost_centre.name, "isFixture": item.is_fixture, } for item in db.query(Account).order_by(Account.type).order_by(Account.name).order_by(Account.code).all() ] @router.get("/query") async def show_term( q: str, t: int = None, r: bool = None, a: bool = None, c: int = None, db: Session = Depends(get_db), current_user: UserToken = Depends(get_user), ): count = c list_ = [] for index, item in enumerate(AccountBase.query(q, t, r, a, db)): list_.append({"id": item.id, "name": item.name}) if count is not None and index == count - 1: break return list_ @router.get("/{id_}/balance") async def show_balance( id_: uuid.UUID, d: str = None, db: Session = Depends(get_db), user: UserToken = Depends(get_user), ): date = None if d is None or d == "" else datetime.strptime(d, "%d-%b-%Y") return {"date": balance(id_, date, db), "total": balance(id_, None, db)} @router.get("/{id_}") def show_id( id_: uuid.UUID, db: Session = Depends(get_db), user: UserToken = Security(get_user, scopes=["accounts"]), ): item: Account = db.query(Account).filter(Account.id == id_).first() return account_info(item) def balance(id_: uuid.UUID, date, db: Session): account = db.query(AccountBase).filter(AccountBase.id == id_).first() if not account.type_object.balance_sheet: return 0 bal = db.query(func.sum(Journal.amount * Journal.debit)).join(Journal.voucher) if date is not None: bal = bal.filter(Voucher.date <= date) bal = bal.filter(Voucher.type != VoucherType.by_name("Issue").id).filter(Journal.account_id == id_).scalar() return 0 if bal is None else bal def account_info(item: Optional[Account]): if item is None: return { "code": "(Auto)", "type": AccountType.by_name("Creditors").id, "isActive": True, "isReconcilable": False, "isStarred": False, "costCentre": CostCentre.overall(), } else: return { "id": item.id, "code": item.code, "name": item.name, "type": item.type, "isActive": item.is_active, "isReconcilable": item.is_reconcilable, "isStarred": item.is_starred, "isFixture": item.is_fixture, "costCentre": {"id": item.cost_centre_id, "name": item.cost_centre.name,}, } def delete_with_data(account: Account, db: Session): suspense_account = db.query(Account).filter(Account.id == Account.suspense()).first() query: List[Voucher] = ( db.query(Voucher) .options(joinedload_all(Voucher.journals, Journal.account, innerjoin=True)) .filter(Voucher.journals.any(Journal.account_id == account.id)) .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: 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.type in (VoucherType.by_name("Payment").id, VoucherType.by_name("Receipt").id,): voucher.type = VoucherType.by_name("Journal") db.delete(account)