barker/barker/barker/routers/settle_option.py

import barker.schemas.settle_option as schemas

from fastapi import APIRouter, Depends, HTTPException, Security, status
from sqlalchemy import func, select
from sqlalchemy.exc import SQLAlchemyError

from ..core.security import get_current_active_user as get_user
from ..db.session import SessionFuture
from ..models.reporting_level import ReportingLevel
from ..models.settle_option import SettleOption
from ..models.voucher_type import VoucherType
from ..schemas.user_token import UserToken


router = APIRouter()


@router.post("", response_model=schemas.SettleOption)
def save(
    data: schemas.SettleOption,
    user: UserToken = Security(get_user, scopes=["owner"]),
) -> schemas.SettleOption:
    try:
        with SessionFuture() as db:
            item = SettleOption(
                id_=db.execute(select(func.max(SettleOption.id) + 1)).scalar_one(),
                name=data.name,
                voucher_type=data.voucher_type,
                reporting_level=data.reporting_level,
                has_reason=data.has_reason,
            )
            db.add(item)
            db.commit()
            return settle_option_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.SettleOption)
def update_route(
    id_: int,
    data: schemas.SettleOption,
    user: UserToken = Security(get_user, scopes=["owner"]),
) -> schemas.SettleOption:
    try:
        with SessionFuture() as db:
            item: SettleOption = db.execute(select(SettleOption).where(SettleOption.id == id_)).scalar_one()
            if item.is_fixture:
                if item.name != data.name:
                    raise HTTPException(
                        status_code=status.HTTP_423_LOCKED,
                        detail=f"{item.name} is a fixture and its name cannot be edited.",
                    )
                if item.voucher_type != data.voucher_type:
                    raise HTTPException(
                        status_code=status.HTTP_423_LOCKED,
                        detail=f"{item.name} is a fixture and its voucher type cannot be edited.",
                    )
                if item.has_reason != data.has_reason:
                    raise HTTPException(
                        status_code=status.HTTP_423_LOCKED,
                        detail=f"{item.name} is a fixture and its has reason property cannot be edited.",
                    )
            else:
                item.name = data.name
                item.voucher_type = data.voucher_type
                item.has_reason = data.has_reason
            item.reporting_level = data.reporting_level
            db.commit()
            return settle_option_info(item)
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


@router.delete("/{id_}")
def delete_route(
    id_: int,
    user: UserToken = Security(get_user, scopes=["owner"]),
):
    with SessionFuture() as db:
        item: SettleOption = db.execute(select(SettleOption).where(SettleOption.id == id_)).scalar_one()
        if item is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Sale Category not found",
            )
        else:
            raise HTTPException(
                status_code=status.HTTP_501_NOT_IMPLEMENTED,
                detail="Sale Category deletion not implemented",
            )


@router.get("", response_model=schemas.SettleOption)
def show_blank(
    user: UserToken = Security(get_user, scopes=["owner"]),
) -> schemas.SettleOption:
    return settle_option_blank()


@router.get("/list", response_model=list[schemas.SettleOption])
def show_list(user: UserToken = Depends(get_user)):
    with SessionFuture() as db:
        return [
            settle_option_info(item)
            for item in db.execute(select(SettleOption).order_by(SettleOption.id)).scalars().all()
        ]


@router.get("/for-type/{vt}", response_model=list[schemas.SettleOption])
def show_list_for_type(vt: VoucherType, user: UserToken = Depends(get_user)):
    with SessionFuture() as db:
        return [
            settle_option_info(item)
            for item in db.execute(
                select(SettleOption).where(SettleOption.voucher_type == vt).order_by(SettleOption.id)
            )
            .scalars()
            .all()
        ]


@router.get("/{id_}", response_model=schemas.SettleOption)
def show_id(
    id_: int,
    user: UserToken = Security(get_user, scopes=["owner"]),
) -> schemas.SettleOption:
    with SessionFuture() as db:
        item: SettleOption = db.execute(select(SettleOption).where(SettleOption.id == id_)).scalar_one()
        return settle_option_info(item)


def settle_option_info(item: SettleOption) -> schemas.SettleOption:
    return schemas.SettleOption(
        id=item.id,
        name=item.name,
        voucherType=item.voucher_type,
        reportingLevel=item.reporting_level,
        hasReason=item.has_reason,
        isFixture=item.is_fixture,
    )


def settle_option_blank() -> schemas.SettleOption:
    return schemas.SettleOption(
        id=0, name="", voucherType=VoucherType.KOT, reportingLevel=ReportingLevel.Skip, hasReason=False, isFixture=False
    )