113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
import uuid
|
|
|
|
import barker.schemas.sale_category 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.product_version import ProductVersion
|
|
from ..models.sale_category import SaleCategory
|
|
from ..schemas.user_token import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_model=schemas.SaleCategory)
|
|
def save(
|
|
data: schemas.SaleCategoryIn,
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.SaleCategory:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item = SaleCategory(name=data.name, discount_limit=round(data.discount_limit, 2), tax_id=data.tax.id_)
|
|
db.add(item)
|
|
db.commit()
|
|
return sale_category_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.SaleCategory)
|
|
def update_route(
|
|
id_: uuid.UUID,
|
|
data: schemas.SaleCategoryIn,
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.SaleCategory:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: SaleCategory = db.execute(select(SaleCategory).where(SaleCategory.id == id_)).scalar_one()
|
|
item.name = data.name
|
|
item.discount_limit = round(data.discount_limit, 2)
|
|
item.tax_id = data.tax.id_
|
|
db.commit()
|
|
return sale_category_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.SaleCategoryBlank)
|
|
def delete_route(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.SaleCategoryBlank:
|
|
with SessionFuture() as db:
|
|
if db.execute(select(count(ProductVersion.id)).where(ProductVersion.sale_category_id == id_)).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Sale Category has active or deleted products and cannot be deleted.",
|
|
)
|
|
item: SaleCategory = db.execute(select(SaleCategory).where(SaleCategory.id == id_)).scalar_one()
|
|
db.delete(item)
|
|
db.commit()
|
|
return sale_category_blank()
|
|
|
|
|
|
@router.get("", response_model=schemas.SaleCategoryBlank)
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.SaleCategoryBlank:
|
|
return sale_category_blank()
|
|
|
|
|
|
@router.get("/list", response_model=list[schemas.SaleCategory])
|
|
def show_list(user: UserToken = Depends(get_user)) -> list[schemas.SaleCategory]:
|
|
with SessionFuture() as db:
|
|
return [
|
|
sale_category_info(item)
|
|
for item in db.execute(select(SaleCategory).order_by(SaleCategory.name)).scalars().all()
|
|
]
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.SaleCategory)
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.SaleCategory:
|
|
with SessionFuture() as db:
|
|
item: SaleCategory = db.execute(select(SaleCategory).where(SaleCategory.id == id_)).scalar_one()
|
|
return sale_category_info(item)
|
|
|
|
|
|
def sale_category_info(item: SaleCategory) -> schemas.SaleCategory:
|
|
return schemas.SaleCategory(
|
|
id=item.id,
|
|
name=item.name,
|
|
discountLimit=item.discount_limit,
|
|
tax=schemas.TaxLink(id=item.tax_id, name=item.tax.name, rate=item.tax.rate),
|
|
)
|
|
|
|
|
|
def sale_category_blank() -> schemas.SaleCategoryBlank:
|
|
return schemas.SaleCategoryBlank(name="", discountLimit=1)
|