tanshu
ca352649f0
1. The Account Types sent were sqlqueries and not integer literals 2. Voucher Types and Account Types were being compared to VoucherTypesSelected and AccountTypesSelected objects and not their ids.
303 lines
10 KiB
Python
303 lines
10 KiB
Python
import uuid
|
|
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from typing import List, Optional
|
|
|
|
import brewman.schemas.input as schema_in
|
|
import brewman.schemas.voucher as output
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Security, status
|
|
from sqlalchemy import and_, distinct, func, or_, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..core.session import get_date, get_first_day, set_date
|
|
from ..db.session import SessionFuture
|
|
from ..models.account import Account
|
|
from ..models.account_base import AccountBase
|
|
from ..models.attendance import Attendance
|
|
from ..models.attendance_type import AttendanceType
|
|
from ..models.employee import Employee
|
|
from ..models.incentive import Incentive
|
|
from ..models.journal import Journal
|
|
from ..models.validations import check_journals_are_valid
|
|
from ..models.voucher import Voucher
|
|
from ..models.voucher_type import VoucherType
|
|
from ..schemas.incentive import Incentive as IncentiveSchema
|
|
from ..schemas.user import UserToken
|
|
from . import get_lock_info
|
|
from .voucher import blank_voucher, check_voucher_edit_allowed, voucher_info
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_model=output.Voucher)
|
|
def save_route(
|
|
request: Request,
|
|
data: schema_in.IncentiveIn = Depends(schema_in.IncentiveIn.load_form),
|
|
user: UserToken = Security(get_user, scopes=["incentive"]),
|
|
):
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Voucher = save(data, user, db)
|
|
employees = get_employees(get_first_day(data.date_), data.date_, data.incentives, None, db)
|
|
amount = balance(data.date_, None, db) * Decimal(0.9) # 10% for Deb Dip
|
|
total_points = sum(e.points * e.days_worked for e in employees)
|
|
point_value = round(amount / total_points, 2)
|
|
save_incentives(item, employees, point_value, db)
|
|
check_journals_are_valid(item)
|
|
db.commit()
|
|
set_date(data.date_.strftime("%d-%b-%Y"), request.session)
|
|
info = voucher_info(item, db)
|
|
return info
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
def save(data: schema_in.IncentiveIn, user: UserToken, db: Session) -> Voucher:
|
|
account_types = (
|
|
db.execute(
|
|
select(distinct(AccountBase.type)).where(AccountBase.id.in_([dj.employee_id for dj in data.incentives]))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
allowed, message = get_lock_info([data.date_], VoucherType.by_name(data.type_).id, account_types, db)
|
|
if not allowed:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_423_LOCKED,
|
|
detail=message,
|
|
)
|
|
voucher = Voucher(
|
|
date=data.date_,
|
|
narration=data.narration,
|
|
is_starred=data.is_starred,
|
|
user_id=user.id_,
|
|
type_=VoucherType.by_name(data.type_),
|
|
)
|
|
db.add(voucher)
|
|
return voucher
|
|
|
|
|
|
def save_incentives(
|
|
voucher: Voucher,
|
|
employees: List[schema_in.IncentiveEmployee],
|
|
point_value: Decimal,
|
|
db: Session,
|
|
):
|
|
total_amount = 0
|
|
for item in employees:
|
|
item_amount = round(item.points * item.days_worked * point_value)
|
|
journal = Journal(
|
|
amount=item_amount,
|
|
debit=-1,
|
|
account_id=item.employee_id,
|
|
cost_centre_id=item.cost_centre_id,
|
|
)
|
|
inc = Incentive(journal=journal, days_worked=item.days_worked, points=item.points)
|
|
voucher.journals.append(journal)
|
|
voucher.incentives.append(inc)
|
|
db.add(journal)
|
|
db.add(inc)
|
|
total_amount += item_amount
|
|
|
|
sc = db.execute(select(Account).where(Account.id == Account.incentive_id())).scalar_one()
|
|
journal = Journal(amount=total_amount, debit=1, account_id=sc.id, cost_centre_id=sc.cost_centre_id)
|
|
voucher.journals.append(journal)
|
|
db.add(journal)
|
|
|
|
|
|
@router.put("/{id_}")
|
|
def update_route(
|
|
id_: uuid.UUID,
|
|
request: Request,
|
|
data: schema_in.IncentiveIn = Depends(schema_in.IncentiveIn.load_form),
|
|
user: UserToken = Security(get_user, scopes=["incentive"]),
|
|
):
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Voucher = update_voucher(id_, data, user, db)
|
|
employees = get_employees(get_first_day(data.date_), data.date_, data.incentives, item.journals, db)
|
|
amount = balance(data.date_, item.id, db) * Decimal(0.9) # 10% for Deb Dip
|
|
total_points = sum(e.points * e.days_worked for e in employees)
|
|
point_value = round(amount / total_points, 2)
|
|
update_incentives(item, employees, point_value, db)
|
|
check_journals_are_valid(item)
|
|
db.commit()
|
|
set_date(data.date_.strftime("%d-%b-%Y"), request.session)
|
|
return voucher_info(item, db)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
def update_voucher(id_: uuid.UUID, data: schema_in.IncentiveIn, user: UserToken, db: Session) -> Voucher:
|
|
voucher: Voucher = db.execute(select(Voucher).where(Voucher.id == id_)).scalar_one()
|
|
account_types = (
|
|
db.execute(
|
|
select(distinct(AccountBase.type)).where(
|
|
AccountBase.id.in_(
|
|
[dj.employee_id for dj in data.incentives] + [vj.account_id for vj in voucher.journals]
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
allowed, message = get_lock_info([voucher.date, data.date_], voucher.type, account_types, db)
|
|
if not allowed:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_423_LOCKED,
|
|
detail=message,
|
|
)
|
|
check_voucher_edit_allowed(voucher, user)
|
|
voucher.is_starred = data.is_starred
|
|
voucher.narration = data.narration
|
|
voucher.user_id = user.id_
|
|
voucher.posted = False
|
|
voucher.last_edit_date = datetime.utcnow()
|
|
return voucher
|
|
|
|
|
|
def update_incentives(
|
|
voucher: Voucher,
|
|
employees: List[schema_in.IncentiveEmployee],
|
|
point_value: Decimal,
|
|
db: Session,
|
|
):
|
|
total_amount = 0
|
|
for item in voucher.incentives:
|
|
employee = next(e for e in employees if e.employee_id == item.journal.account_id)
|
|
item_amount = round(employee.points * employee.days_worked * point_value)
|
|
item.days_worked = employee.days_worked
|
|
item.points = employee.points
|
|
item.journal.amount = item_amount
|
|
total_amount += item_amount
|
|
|
|
journal = next(j for j in voucher.journals if j.account_id == Account.incentive_id())
|
|
journal.amount = total_amount
|
|
|
|
|
|
@router.get("/{id_}", response_model=output.Voucher)
|
|
def get_id(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["incentive"]),
|
|
):
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Voucher = db.execute(select(Voucher).where(Voucher.id == id_)).scalar_one()
|
|
return voucher_info(item, db)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=output.Voucher)
|
|
def show_blank(
|
|
request: Request,
|
|
d: str = None,
|
|
user: UserToken = Security(get_user, scopes=["incentive"]),
|
|
):
|
|
additional_info = {"date": d or get_date(request.session), "type": "Incentive"}
|
|
with SessionFuture() as db:
|
|
return blank_voucher(additional_info, db)
|
|
|
|
|
|
def get_employees(
|
|
start_date: date,
|
|
finish_date: date,
|
|
incentives: List[IncentiveSchema],
|
|
journals: Optional[List[Journal]],
|
|
db: Session,
|
|
) -> List[schema_in.IncentiveEmployee]:
|
|
details = []
|
|
employees = db.execute(
|
|
select(Employee)
|
|
.where(Employee.joining_date <= finish_date, or_(Employee.is_active, Employee.leaving_date >= start_date))
|
|
.order_by(Employee.cost_centre_id, Employee.designation, Employee.name)
|
|
.all()
|
|
)
|
|
|
|
check_if_employees_changed(incentives, employees, journals)
|
|
|
|
for employee in employees:
|
|
att = (
|
|
db.execute(
|
|
select(Attendance).where(
|
|
Attendance.employee_id == employee.id,
|
|
Attendance.date >= start_date,
|
|
Attendance.date <= finish_date,
|
|
Attendance.is_valid == True, # noqa: E712
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
att = 0.5 * round(sum(map(lambda x: AttendanceType.by_id(x.attendance_type).value, att)) / 0.5)
|
|
points = next(x for x in incentives if x.employee_id == employee.id).points
|
|
details.append(
|
|
schema_in.IncentiveEmployee(
|
|
employee_id=employee.id,
|
|
cost_centre_id=employee.cost_centre_id,
|
|
days_worked=att,
|
|
points=points,
|
|
)
|
|
)
|
|
return details
|
|
|
|
|
|
def balance(date_: date, voucher_id: Optional[uuid.UUID], db: Session):
|
|
amount = db.execute(
|
|
select(func.sum(Journal.amount * Journal.debit))
|
|
.join(Journal.voucher)
|
|
.where(
|
|
Journal.account_id == Account.incentive_id(),
|
|
Voucher.type != VoucherType.by_name("Issue").id,
|
|
or_(
|
|
Voucher.date <= date_,
|
|
and_(
|
|
Voucher.date == date_,
|
|
Voucher.type != VoucherType.by_name("Incentive").id,
|
|
),
|
|
),
|
|
)
|
|
)
|
|
if voucher_id is not None:
|
|
amount = amount.where(Voucher.id != voucher_id)
|
|
result: Decimal = amount.scalar()
|
|
return 0 if result is None else result * -1
|
|
|
|
|
|
def check_if_employees_changed(
|
|
json: List[IncentiveSchema],
|
|
db: List[Employee],
|
|
voucher: Optional[List[Journal]],
|
|
):
|
|
json = set(x.employee_id for x in json)
|
|
db = set(x.id for x in db)
|
|
voucher = (
|
|
set(x.account_id for x in voucher if x.account_id != Account.incentive_id()) if voucher is not None else None
|
|
)
|
|
if voucher is None:
|
|
if len(json ^ db) != 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Employee missing in json data",
|
|
)
|
|
else:
|
|
if len(json ^ db) != 0 or len(db ^ voucher) != 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Employee missing in json data",
|
|
)
|