257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
import uuid
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Sequence
|
|
|
|
import brewman.schemas.employee as schemas
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import desc, 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.cost_centre import CostCentre
|
|
from ..models.employee import Employee
|
|
from ..models.journal import Journal
|
|
from ..models.voucher import Voucher
|
|
from ..models.voucher_type import VoucherType
|
|
from ..schemas.user import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_model=schemas.Employee)
|
|
def save(
|
|
data: schemas.EmployeeIn,
|
|
user: UserToken = Security(get_user, scopes=["employees"]),
|
|
) -> schemas.Employee:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item = Employee(
|
|
name=data.name,
|
|
code=None,
|
|
is_starred=data.is_starred,
|
|
is_active=data.is_active,
|
|
cost_centre_id=data.cost_centre.id_,
|
|
designation=data.designation,
|
|
salary=data.salary,
|
|
points=data.points,
|
|
joining_date=data.joining_date,
|
|
leaving_date=None if data.is_active else data.leaving_date,
|
|
).create(db)
|
|
db.commit()
|
|
return employee_info(item)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.put("/{id_}", response_model=schemas.Employee)
|
|
def update_route(
|
|
id_: uuid.UUID,
|
|
data: schemas.EmployeeIn,
|
|
user: UserToken = Security(get_user, scopes=["employees"]),
|
|
) -> schemas.Employee:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Employee = db.execute(select(Employee).where(Employee.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.",
|
|
)
|
|
item.name = data.name
|
|
item.cost_centre_id = data.cost_centre.id_
|
|
item.designation = data.designation
|
|
item.salary = data.salary
|
|
item.points = data.points
|
|
item.joining_date = data.joining_date
|
|
item.is_starred = data.is_starred
|
|
item.is_active = data.is_active
|
|
item.leaving_date = data.leaving_date
|
|
db.commit()
|
|
return employee_info(item)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.delete("/{id_}", response_model=schemas.EmployeeBlank)
|
|
def delete_route(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["employees"]),
|
|
) -> schemas.EmployeeBlank:
|
|
with SessionFuture() as db:
|
|
employee: Employee = db.execute(select(Employee).where(Employee.id == id_)).scalar_one()
|
|
can_delete, reason = employee.can_delete("advanced-delete" in user.permissions)
|
|
if can_delete:
|
|
delete_with_data(employee, db)
|
|
db.commit()
|
|
return employee_blank()
|
|
else:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Cannot delete account because {reason}",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=schemas.EmployeeBlank)
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["employees"]),
|
|
) -> schemas.EmployeeBlank:
|
|
return employee_blank()
|
|
|
|
|
|
@router.get("/list", response_model=list[schemas.Employee])
|
|
async def show_list(user: UserToken = Depends(get_user)) -> list[schemas.Employee]:
|
|
with SessionFuture() as db:
|
|
return [
|
|
employee_info(item)
|
|
for item in db.execute(
|
|
select(Employee)
|
|
.order_by(desc(Employee.is_active))
|
|
.order_by(Account.cost_centre_id)
|
|
.order_by(Employee.designation)
|
|
.order_by(Employee.name)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
]
|
|
|
|
|
|
@router.get("/query", response_model=list[schemas.EmployeeLink])
|
|
async def show_term(
|
|
q: str,
|
|
c: int | None = None,
|
|
current_user: UserToken = Depends(get_user),
|
|
) -> list[schemas.EmployeeLink]:
|
|
list_: list[schemas.EmployeeLink] = []
|
|
with SessionFuture() as db:
|
|
query_ = select(Employee)
|
|
if q is not None:
|
|
for name in q.split():
|
|
query_ = query_.where(Employee.name.ilike(f"%{name}%"))
|
|
query_ = query_.order_by(Employee.name)
|
|
if c is not None:
|
|
query_ = query_.limit(c)
|
|
data: Sequence[Employee] = db.execute(query_).scalars().all()
|
|
|
|
for item in data:
|
|
list_.append(
|
|
schemas.EmployeeLink(
|
|
id_=item.id,
|
|
name=item.name,
|
|
designation=item.designation,
|
|
cost_centre=schemas.CostCentreLink(
|
|
id_=item.cost_centre.id,
|
|
name=item.cost_centre.name,
|
|
),
|
|
)
|
|
)
|
|
return list_
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.Employee)
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["employees"]),
|
|
) -> schemas.Employee:
|
|
with SessionFuture() as db:
|
|
item: Employee = db.execute(select(Employee).where(Employee.id == id_)).scalar_one()
|
|
return employee_info(item)
|
|
|
|
|
|
def employee_info(employee: Employee) -> schemas.Employee:
|
|
return schemas.Employee(
|
|
id_=employee.id,
|
|
code=employee.code,
|
|
name=employee.name,
|
|
is_active=employee.is_active,
|
|
is_starred=employee.is_starred,
|
|
designation=employee.designation,
|
|
salary=employee.salary,
|
|
points=employee.points,
|
|
joining_date=employee.joining_date,
|
|
leaving_date=None if employee.is_active else employee.leaving_date,
|
|
cost_centre=schemas.CostCentreLink(
|
|
id_=employee.cost_centre_id,
|
|
name=employee.cost_centre.name,
|
|
),
|
|
is_fixture=employee.is_fixture,
|
|
)
|
|
|
|
|
|
def employee_blank() -> schemas.EmployeeBlank:
|
|
return schemas.EmployeeBlank(
|
|
name="",
|
|
is_starred=False,
|
|
is_active=True,
|
|
cost_centre=CostCentre.overall(),
|
|
designation="",
|
|
salary=0,
|
|
points=Decimal(0),
|
|
joining_date=datetime.today().date(),
|
|
leaving_date=None,
|
|
)
|
|
|
|
|
|
def delete_with_data(employee: Employee, db: Session) -> None:
|
|
suspense_account = db.execute(select(Account).where(Account.id == Account.suspense())).scalar_one()
|
|
query = (
|
|
db.execute(
|
|
select(Voucher)
|
|
.options(joinedload(Voucher.journals, innerjoin=True).joinedload(Journal.account, innerjoin=True))
|
|
.where(Voucher.journals.any(Journal.account_id == employee.id))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
for voucher in query:
|
|
others, sus_jnl, acc_jnl = False, None, None
|
|
for journal in voucher.journals:
|
|
if journal.account_id == employee.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 {employee.name}"
|
|
else:
|
|
amount = (sus_jnl.debit * sus_jnl.amount) + (acc_jnl.debit * acc_jnl.amount)
|
|
if acc_jnl.employee_benefit is not None:
|
|
db.delete(acc_jnl.employee_benefit)
|
|
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 {employee.name}"
|
|
if voucher.voucher_type in (
|
|
VoucherType.PAYMENT,
|
|
VoucherType.RECEIPT,
|
|
):
|
|
voucher.voucher_type = VoucherType.JOURNAL
|
|
for fingerprint in employee.fingerprints:
|
|
db.delete(fingerprint)
|
|
for attendance in employee.attendances:
|
|
db.delete(attendance)
|
|
db.delete(employee)
|