barker/barker/barker/routers/tax.py

import re
import uuid

from decimal import Decimal
from typing import List

import barker.schemas.tax as schemas

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

from ..core.security import get_current_active_user as get_user
from ..db.session import SessionLocal
from ..models.master import SaleCategory, Tax
from ..models.voucher import Inventory
from ..schemas.auth import UserToken


router = APIRouter()


# Dependency
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@router.post("", response_model=schemas.Tax)
def save(
    data: schemas.TaxIn,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.Tax:
    try:
        item = Tax(name=data.name, rate=round(data.rate, 5))
        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:
        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.Tax)
def update(
    id_: uuid.UUID,
    data: schemas.TaxIn,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.Tax:
    try:
        item: Tax = db.query(Tax).filter(Tax.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 name_valid(data.name):
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail="The name is not valid",
            )
        item.name = data.name
        item.rate = round(data.rate, 5)
        db.commit()
        return tax_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_}", response_model=schemas.TaxBlank)
def delete(
    id_: uuid.UUID,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.TaxBlank:
    try:
        item: Tax = db.query(Tax).filter(Tax.id == id_).first()
        if item is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Tax not found",
            )
        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.query(count(SaleCategory.tax_id)).filter(SaleCategory.tax_id == item.id).scalar() > 0:
            raise HTTPException(
                status_code=status.HTTP_423_LOCKED,
                detail=f"{item.name} has associated Sale Categories and cannot be deleted",
            )
        if db.query(count(Inventory.tax_id)).filter(Inventory.tax_id == item.id).scalar() > 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()
    except Exception:
        db.rollback()
        raise


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


@router.get("/list", response_model=List[schemas.Tax])
def show_list(db: Session = Depends(get_db), user: UserToken = Depends(get_user)) -> List[schemas.Tax]:
    return [tax_info(item) for item in db.query(Tax).order_by(Tax.name).all()]


@router.get("/{id_}", response_model=schemas.Tax)
def show_id(
    id_: uuid.UUID,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["taxes"]),
) -> schemas.Tax:
    item: Tax = db.query(Tax).filter(Tax.id == id_).first()
    return tax_info(item)


def tax_info(item: Tax) -> schemas.Tax:
    return schemas.Tax(
        id=item.id,
        name=item.name,
        rate=item.rate,
        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 = 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