530 lines
19 KiB
Python
530 lines
19 KiB
Python
import uuid
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import Date, func, or_, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import contains_eager
|
|
|
|
from ..core.config import settings
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.bundle_item import BundleItem as BundleItemModel
|
|
from ..models.inventory import Inventory
|
|
from ..models.kot import Kot
|
|
from ..models.product import Product
|
|
from ..models.product_version import ProductVersion
|
|
from ..models.sku_version import SkuVersion
|
|
from ..models.stock_keeping_unit import StockKeepingUnit
|
|
from ..models.voucher import Voucher
|
|
from ..schemas import bundle as schemas
|
|
from ..schemas.menu_category import MenuCategoryLink
|
|
from ..schemas.sale_category import SaleCategoryLink
|
|
from ..schemas.user_token import UserToken
|
|
from . import _pv_onclause, _sv_onclause, effective_date
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
FRACTION_1 = Decimal("1")
|
|
YIELD_1 = Decimal("1")
|
|
COST_0 = Decimal("0")
|
|
|
|
|
|
# ---------- helpers ----------
|
|
|
|
|
|
def bundle_blank() -> schemas.BundleBlank:
|
|
return schemas.BundleBlank(
|
|
name="",
|
|
units="",
|
|
sale_price=Decimal("0.00"),
|
|
sale_category=None,
|
|
has_happy_hour=False,
|
|
is_not_available=False,
|
|
sort_order=0,
|
|
items=[],
|
|
)
|
|
|
|
|
|
def _bundle_info(pv: ProductVersion, sv: SkuVersion, items: Sequence[BundleItemModel]) -> schemas.Bundle:
|
|
sku = sv.sku
|
|
return schemas.Bundle(
|
|
id_=sku.id,
|
|
version_id=sv.id,
|
|
name=pv.name,
|
|
units=pv.fraction_units,
|
|
sale_price=sv.sale_price,
|
|
sale_category=SaleCategoryLink(id_=pv.sale_category_id, name=pv.sale_category.name),
|
|
has_happy_hour=sv.has_happy_hour,
|
|
is_not_available=sku.is_not_available,
|
|
sort_order=sku.sort_order,
|
|
menu_category=MenuCategoryLink(
|
|
id_=sv.menu_category_id,
|
|
name=sv.menu_category.name,
|
|
skus=[],
|
|
)
|
|
if sv.menu_category is not None
|
|
else None,
|
|
items=[
|
|
schemas.BundleItem(
|
|
id_=bi.id,
|
|
name=f"{bi.item.product.versions[0].name} ({bi.item.versions[0].units})",
|
|
item_id=bi.item_id,
|
|
sale_price=bi.sale_price,
|
|
quantity=bi.quantity,
|
|
print_in_bill=bi.print_in_bill,
|
|
)
|
|
for bi in items
|
|
],
|
|
)
|
|
|
|
|
|
# ---------- routes ----------
|
|
|
|
|
|
@router.post("", response_model=None)
|
|
def save(
|
|
data: schemas.BundleIn,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
if not data.items:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Not enough bundle items.",
|
|
)
|
|
sale_price = round(
|
|
sum((round(it.sale_price, 2) * round(it.quantity, 5) for it in data.items), start=Decimal(0)), 2
|
|
)
|
|
|
|
product = Product(sort_order=data.sort_order)
|
|
db.add(product)
|
|
db.flush()
|
|
|
|
pv = ProductVersion(
|
|
product_id=None,
|
|
product=product,
|
|
name=data.name,
|
|
fraction_units=data.units, # per your rule
|
|
sale_category_id=data.sale_category.id_,
|
|
sale_category=None,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
db.add(pv)
|
|
|
|
sku = StockKeepingUnit(
|
|
product=product,
|
|
is_not_available=data.is_not_available,
|
|
is_bundle=True,
|
|
sort_order=data.sort_order,
|
|
)
|
|
db.add(sku)
|
|
db.flush()
|
|
|
|
sv = SkuVersion(
|
|
sku_id=sku.id,
|
|
units=data.units,
|
|
fraction=FRACTION_1,
|
|
product_yield=YIELD_1,
|
|
cost_price=COST_0,
|
|
sale_price=sale_price,
|
|
has_happy_hour=data.has_happy_hour,
|
|
menu_category_id=data.menu_category.id_,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
db.add(sv)
|
|
|
|
# bundle items
|
|
for it in data.items:
|
|
db.add(
|
|
BundleItemModel(
|
|
bundle_id=sku.id,
|
|
item_id=it.item_id,
|
|
quantity=round(it.quantity, 5),
|
|
sale_price=round(it.sale_price, 2),
|
|
print_in_bill=it.print_in_bill,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
)
|
|
|
|
db.commit()
|
|
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
|
|
@router.put("/{id_}", response_model=None)
|
|
def update_route(
|
|
id_: uuid.UUID, # bundle header SKU id
|
|
data: schemas.BundleIn,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
if not data.items:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Not enough bundle items.",
|
|
)
|
|
sale_price = round(
|
|
sum((round(it.sale_price, 2) * round(it.quantity, 5) for it in data.items), start=Decimal(0)), 2
|
|
)
|
|
# Load header SKU + active header SkuVersion + menu category + product + active ProductVersion
|
|
sv: SkuVersion | None = (
|
|
db.execute(
|
|
select(SkuVersion)
|
|
.join(StockKeepingUnit, onclause=_sv_onclause(date_))
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=_pv_onclause(date_))
|
|
.join(SkuVersion.menu_category)
|
|
.where(
|
|
StockKeepingUnit.id == id_,
|
|
StockKeepingUnit.is_bundle == True, # noqa: E712
|
|
)
|
|
.options(
|
|
contains_eager(SkuVersion.sku)
|
|
.contains_eager(StockKeepingUnit.product)
|
|
.contains_eager(Product.versions),
|
|
contains_eager(SkuVersion.menu_category),
|
|
)
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.one_or_none()
|
|
)
|
|
|
|
if sv is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Bundle not found.")
|
|
|
|
sku = sv.sku
|
|
product = sku.product
|
|
pv = product.versions[0]
|
|
pv_changed = pv.name != data.name or pv.sale_category_id != data.sale_category.id_
|
|
|
|
sku.is_not_available = data.is_not_available
|
|
sku.sort_order = data.sort_order
|
|
product.sort_order = data.sort_order
|
|
|
|
if pv_changed:
|
|
if pv.valid_from == date_:
|
|
pv.name = data.name
|
|
pv.sale_category_id = data.sale_category.id_
|
|
else:
|
|
pv.valid_till = date_ - timedelta(days=1)
|
|
db.add(
|
|
ProductVersion(
|
|
product_id=product.id,
|
|
product=product,
|
|
name=data.name,
|
|
fraction_units=pv.fraction_units,
|
|
sale_category_id=data.sale_category.id_,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
)
|
|
sv_changed = (
|
|
sv.units != data.units
|
|
or Decimal(sv.sale_price).quantize(Decimal("0.01")) != sale_price
|
|
or sv.has_happy_hour != data.has_happy_hour
|
|
or sv.menu_category_id != data.menu_category.id_
|
|
)
|
|
if sv_changed:
|
|
if sv.valid_from == date_:
|
|
sv.units = data.units
|
|
sv.sale_price = sale_price
|
|
sv.has_happy_hour = data.has_happy_hour
|
|
sv.menu_category_id = data.menu_category.id_
|
|
|
|
# # enforce bundle constants
|
|
# sv.fraction = FRACTION_1
|
|
# sv.product_yield = YIELD_1
|
|
# sv.cost_price = COST_0
|
|
else:
|
|
sv.valid_till = date_ - timedelta(days=1)
|
|
db.add(
|
|
SkuVersion(
|
|
sku_id=sku.id,
|
|
units=data.units,
|
|
fraction=FRACTION_1,
|
|
product_yield=YIELD_1,
|
|
cost_price=COST_0,
|
|
sale_price=sale_price,
|
|
has_happy_hour=data.has_happy_hour,
|
|
menu_category_id=data.menu_category.id_,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
)
|
|
|
|
# ---- Bundle items replace/update
|
|
existing = db.execute(select(BundleItemModel).where(BundleItemModel.bundle_id == sku.id)).scalars().all()
|
|
existing_by_item = {x.item_id: x for x in existing}
|
|
incoming_ids = {x.item_id for x in data.items}
|
|
|
|
existing_by_item = {x.item_id: x for x in existing}
|
|
|
|
incoming_ids = {x.item_id for x in data.items}
|
|
|
|
# delete removed
|
|
for ex_d in existing:
|
|
if ex_d.item_id not in incoming_ids:
|
|
if ex_d.valid_from == date_:
|
|
db.delete(ex_d)
|
|
else:
|
|
ex_d.valid_till = date_ - timedelta(days=1)
|
|
|
|
# add/update
|
|
for it in data.items:
|
|
ex = existing_by_item.get(it.item_id)
|
|
if ex is None:
|
|
db.add(
|
|
BundleItemModel(
|
|
bundle_id=sku.id,
|
|
item_id=it.item_id,
|
|
sale_price=round(it.sale_price, 2),
|
|
quantity=round(it.quantity, 5),
|
|
print_in_bill=it.print_in_bill,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
)
|
|
elif ex.valid_from == date_:
|
|
ex.sale_price = round(it.sale_price, 2)
|
|
ex.quantity = round(it.quantity, 5)
|
|
else:
|
|
ex.valid_till = date_ - timedelta(days=1)
|
|
db.add(
|
|
BundleItemModel(
|
|
bundle_id=sku.id,
|
|
item_id=it.item_id,
|
|
sale_price=round(it.sale_price, 2),
|
|
quantity=round(it.quantity, 5),
|
|
print_in_bill=it.print_in_bill,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
)
|
|
|
|
db.commit()
|
|
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
|
|
@router.get("", response_model=schemas.BundleBlank)
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.BundleBlank:
|
|
return bundle_blank()
|
|
|
|
|
|
@router.get("/list", response_model=list[schemas.Bundle])
|
|
def show_list(
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> list[schemas.Bundle]:
|
|
with SessionFuture() as db:
|
|
sv_onclause = _sv_onclause(date_)
|
|
pv_onclause = _pv_onclause(date_)
|
|
|
|
rows = (
|
|
db.execute(
|
|
select(SkuVersion)
|
|
# .select_from(SkuVersion)
|
|
.join(StockKeepingUnit, onclause=sv_onclause)
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=pv_onclause)
|
|
.join(SkuVersion.menu_category)
|
|
.where(StockKeepingUnit.is_bundle == True) # noqa: E712
|
|
.options(
|
|
contains_eager(SkuVersion.sku).contains_eager(StockKeepingUnit.product),
|
|
contains_eager(SkuVersion.menu_category),
|
|
)
|
|
.order_by(StockKeepingUnit.sort_order, SkuVersion.units)
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
out: list[schemas.Bundle] = []
|
|
for sv in rows:
|
|
pv = sv.sku.product.versions[0]
|
|
items = (
|
|
db.execute(
|
|
select(BundleItemModel)
|
|
.join(BundleItemModel.item)
|
|
.join(SkuVersion, onclause=sv_onclause)
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=pv_onclause)
|
|
.join(ProductVersion.sale_category)
|
|
.where(
|
|
BundleItemModel.bundle_id == sv.sku_id,
|
|
or_(BundleItemModel.valid_from == None, BundleItemModel.valid_from <= date_), # noqa: E711
|
|
or_(BundleItemModel.valid_till == None, BundleItemModel.valid_till >= date_), # noqa: E711
|
|
)
|
|
.options(
|
|
contains_eager(BundleItemModel.item).contains_eager(StockKeepingUnit.versions),
|
|
contains_eager(BundleItemModel.item)
|
|
.contains_eager(StockKeepingUnit.product)
|
|
.contains_eager(Product.versions)
|
|
.contains_eager(ProductVersion.sale_category),
|
|
)
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.all()
|
|
)
|
|
out.append(_bundle_info(pv=pv, sv=sv, items=items))
|
|
return out
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.Bundle)
|
|
def show_id(
|
|
id_: uuid.UUID, # bundle header SKU id
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.Bundle:
|
|
with SessionFuture() as db:
|
|
sv_onclause = _sv_onclause(date_)
|
|
pv_onclause = _pv_onclause(date_)
|
|
|
|
sv = (
|
|
db.execute(
|
|
select(SkuVersion)
|
|
.join(StockKeepingUnit, onclause=sv_onclause)
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=pv_onclause)
|
|
.join(SkuVersion.menu_category)
|
|
.where(
|
|
StockKeepingUnit.id == id_,
|
|
StockKeepingUnit.is_bundle == True, # noqa: E712
|
|
)
|
|
.options(
|
|
contains_eager(SkuVersion.sku).contains_eager(StockKeepingUnit.product),
|
|
contains_eager(SkuVersion.menu_category),
|
|
)
|
|
)
|
|
.unique()
|
|
.scalar_one_or_none()
|
|
)
|
|
|
|
if sv is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Bundle not found.")
|
|
|
|
pv = sv.sku.product.versions[0]
|
|
|
|
items = (
|
|
db.execute(
|
|
select(BundleItemModel)
|
|
.join(BundleItemModel.item)
|
|
.join(SkuVersion, onclause=sv_onclause)
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=pv_onclause)
|
|
.join(ProductVersion.sale_category)
|
|
.where(
|
|
BundleItemModel.bundle_id == sv.sku_id,
|
|
or_(BundleItemModel.valid_from == None, BundleItemModel.valid_from <= date_), # noqa: E711
|
|
or_(BundleItemModel.valid_till == None, BundleItemModel.valid_till >= date_), # noqa: E711
|
|
)
|
|
.options(
|
|
contains_eager(BundleItemModel.item).contains_eager(StockKeepingUnit.versions),
|
|
contains_eager(BundleItemModel.item)
|
|
.contains_eager(StockKeepingUnit.product)
|
|
.contains_eager(Product.versions)
|
|
.contains_eager(ProductVersion.sale_category),
|
|
)
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return _bundle_info(pv=pv, sv=sv, items=items)
|
|
|
|
|
|
@router.delete("/{id_}", response_model=None)
|
|
def delete_route(
|
|
id_: uuid.UUID, # bundle header SKU id
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
with SessionFuture() as db:
|
|
sv_onclause = _sv_onclause(date_)
|
|
pv_onclause = _pv_onclause(date_)
|
|
day = func.cast(
|
|
Voucher.date + timedelta(minutes=settings.TIMEZONE_OFFSET_MINUTES - settings.NEW_DAY_OFFSET_MINUTES), Date
|
|
).label("day")
|
|
billed = db.execute(
|
|
select(func.count(Inventory.id))
|
|
.join(Inventory.kot)
|
|
.join(Kot.voucher)
|
|
.where(Inventory.sku_id == id_, day >= date_)
|
|
).scalar_one()
|
|
if billed > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="The cannot delete this product as it was billed",
|
|
)
|
|
|
|
sv = (
|
|
db.execute(
|
|
select(SkuVersion)
|
|
.join(StockKeepingUnit, onclause=sv_onclause)
|
|
.join(StockKeepingUnit.product)
|
|
.join(ProductVersion, onclause=pv_onclause)
|
|
.where(
|
|
StockKeepingUnit.id == id_,
|
|
StockKeepingUnit.is_bundle == True, # noqa: E712
|
|
)
|
|
.options(contains_eager(SkuVersion.sku).contains_eager(StockKeepingUnit.product))
|
|
)
|
|
.unique()
|
|
.scalar_one_or_none()
|
|
)
|
|
|
|
if sv is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Bundle not found.")
|
|
|
|
pv = sv.sku.product.versions[0]
|
|
|
|
# close/delete sku version
|
|
if sv.valid_from == date_:
|
|
db.delete(sv)
|
|
else:
|
|
sv.valid_till = date_ - timedelta(days=1)
|
|
|
|
# close/delete product version
|
|
if pv.valid_from == date_:
|
|
db.delete(pv)
|
|
else:
|
|
pv.valid_till = date_ - timedelta(days=1)
|
|
|
|
items = (
|
|
db.execute(
|
|
select(BundleItemModel).where(
|
|
BundleItemModel.bundle_id == id_,
|
|
or_(BundleItemModel.valid_from == None, BundleItemModel.valid_from <= date_), # noqa: E711
|
|
or_(BundleItemModel.valid_till == None, BundleItemModel.valid_till >= date_), # noqa: E711
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for bi in items:
|
|
if bi.valid_from == date_:
|
|
db.delete(bi)
|
|
else:
|
|
bi.valid_till = date_ - timedelta(days=1)
|
|
|
|
db.commit()
|