barker/barker/barker/routers/tax.py

import re
import uuid

from decimal import Decimal

import barker.schemas.tax as schemas

from fastapi import APIRouter, Depends, HTTPException, Security, status
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.sql.functions import count

from ..core.security import get_current_active_user as get_user
from ..db.session import SessionFuture
from ..models.inventory import Inventory
from ..models.sale_category import SaleCategory
from ..models.tax import Tax
from ..schemas.user_token import UserToken


router = APIRouter()


@router.post("", response_model=schemas.Tax)
def save(
    data: schemas.TaxIn,
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.Tax:
    try:
        with SessionFuture() as db:
            item = Tax(name=data.name, rate=round(data.rate, 5), regime_id=data.regime.id_)
            if not name_valid(data.name):
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail="The name is not valid",
                )
            db.add(item)
            db.commit()
            return tax_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.Tax)
def update_route(
    id_: uuid.UUID,
    data: schemas.TaxIn,
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.Tax:
    try:
        with SessionFuture() as db:
            item: Tax = db.execute(select(Tax).where(Tax.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 name_valid(data.name):
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail="The name is not valid",
                )
            item.regime_id = data.regime.id_
            item.name = data.name
            item.rate = round(data.rate, 5)
            db.commit()
            return tax_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.TaxBlank)
def delete_route(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.TaxBlank:
    with SessionFuture() as db:
        item: Tax = db.execute(select(Tax).where(Tax.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 db.execute(select(count(SaleCategory.tax_id)).where(SaleCategory.tax_id == item.id)).scalar_one() > 0:
            raise HTTPException(
                status_code=status.HTTP_423_LOCKED,
                detail=f"{item.name} has associated Sale Categories and cannot be deleted",
            )
        if db.execute(select(count(Inventory.tax_id)).where(Inventory.tax_id == item.id)).scalar_one() > 0:
            raise HTTPException(
                status_code=status.HTTP_423_LOCKED,
                detail=f"{item.name} has associated Inventories and cannot be deleted",
            )
        db.delete(item)
        db.commit()
        return tax_blank()


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


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


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


def tax_info(item: Tax) -> schemas.Tax:
    return schemas.Tax(
        id=item.id,
        name=item.name,
        rate=item.rate,
        regime=schemas.RegimeLink(id=item.regime_id, name=item.regime.name),
        isFixture=item.is_fixture,
    )


def tax_blank() -> schemas.TaxBlank:
    return schemas.TaxBlank(name="", rate=0, isFixture=False)


def name_valid(name: str) -> bool:
    items = name.split(";")
    if len(items) == 1:
        return True
    total = Decimal(0)
    for i, item in enumerate(it.strip() for it in items):
        match = re.match(r"(^.*)\s+\((.*?)/(.*?)\)[^(]*$", item)
        if not match or len(match.groups()) != 3:
            return False
        total += round(Decimal(match.group(2)) / Decimal(match.group(3)), 5)
    if round(total, 2) != 1:
        return False
    return True