258 lines
10 KiB
Python
258 lines
10 KiB
Python
import uuid
|
|
|
|
from datetime import date, timedelta
|
|
from typing import Dict, List
|
|
|
|
import barker.schemas.product as schemas
|
|
|
|
from fastapi import APIRouter, HTTPException, Security, status
|
|
from sqlalchemy import and_, delete, distinct, or_, select, update
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session, contains_eager, joinedload
|
|
from sqlalchemy.sql.functions import count, func
|
|
|
|
from ..core.config import settings
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.inventory import Inventory
|
|
from ..models.kot import Kot
|
|
from ..models.menu_category import MenuCategory
|
|
from ..models.product import Product
|
|
from ..models.product_version import ProductVersion
|
|
from ..models.sale_category import SaleCategory
|
|
from ..models.voucher import Voucher
|
|
from ..schemas.user_token import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.put("/{version_id}", response_model=None)
|
|
def update_route(
|
|
version_id: uuid.UUID,
|
|
data: schemas.Product,
|
|
user: UserToken = Security(get_user, scopes=["temporal-products"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
old: ProductVersion = db.execute(select(ProductVersion).where(ProductVersion.id == version_id)).scalar_one()
|
|
if (
|
|
old.product_id != data.id_
|
|
or old.name != data.name
|
|
or old.units != data.units
|
|
or old.valid_from != data.valid_from
|
|
or old.valid_till != data.valid_till
|
|
):
|
|
check_product(old, data, db)
|
|
check_inventories(old, data, db)
|
|
update_inventories(old, data, db)
|
|
old.product_id = data.id_
|
|
old.name = data.name
|
|
old.units = data.units
|
|
old.menu_category_id = data.menu_category.id_
|
|
old.sale_category_id = data.sale_category.id_
|
|
old.price = data.price
|
|
old.has_happy_hour = data.has_happy_hour
|
|
old.is_not_available = data.is_not_available
|
|
old.quantity = data.quantity
|
|
old.valid_from = data.valid_from
|
|
old.valid_till = data.valid_till
|
|
db.flush()
|
|
db.execute(
|
|
delete(Product)
|
|
.where(~Product.id.in_(select(distinct(ProductVersion.product_id))))
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
db.commit()
|
|
return
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
def check_product(old: ProductVersion, data: schemas.Product, db: Session):
|
|
query = select(count(ProductVersion.id)).where(ProductVersion.id != old.id)
|
|
if data.valid_from is not None:
|
|
query = query.where(ProductVersion.valid_till >= data.valid_from)
|
|
if data.valid_till is not None:
|
|
query = query.where(ProductVersion.valid_from <= data.valid_till)
|
|
query = query.where(
|
|
or_(
|
|
ProductVersion.product_id == data.id_,
|
|
and_(ProductVersion.name == data.name, ProductVersion.units == data.units),
|
|
)
|
|
)
|
|
if db.execute(query).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Overlapping product exists",
|
|
)
|
|
|
|
|
|
def check_inventories(old: ProductVersion, data: schemas.Product, db: Session):
|
|
day = func.date_trunc(
|
|
"day", Voucher.date + timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES - settings.TIMEZONE_OFFSET_MINUTES)
|
|
).label("day")
|
|
|
|
if data.valid_from is not None and (old.valid_from or date.min) < data.valid_from:
|
|
query = select(count(Inventory.id)).where(Inventory.product_id == old.product_id)
|
|
if old.valid_from is not None:
|
|
query = query.where(day >= old.valid_from)
|
|
query = query.where(day < data.valid_from)
|
|
if db.execute(query).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Changing of validity will orphan inventories",
|
|
)
|
|
|
|
if data.valid_till is not None and (old.valid_till or date.max) > data.valid_till:
|
|
query = select(count(Inventory.id)).where(Inventory.product_id == old.product_id)
|
|
if old.valid_till is not None:
|
|
query = query.where(day <= old.valid_till)
|
|
query = query.where(day > data.valid_till)
|
|
if db.execute(query).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Changing of validity will orphan inventories",
|
|
)
|
|
|
|
|
|
def update_inventories(old: ProductVersion, data: schemas.Product, db: Session):
|
|
if old.product_id != data.id_:
|
|
day = func.date_trunc(
|
|
"day", Voucher.date + timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES - settings.TIMEZONE_OFFSET_MINUTES)
|
|
).label("day")
|
|
invs = select(Inventory.id).join(Inventory.kot).join(Kot.voucher).where(Inventory.product_id == old.product_id)
|
|
if old.valid_from is not None:
|
|
invs = invs.where(day >= old.valid_from)
|
|
if old.valid_till is not None:
|
|
invs = invs.where(day <= old.valid_till)
|
|
db.execute(
|
|
update(Inventory)
|
|
.values(product_id=data.id_)
|
|
.where(Inventory.id.in_(invs))
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
|
|
|
|
@router.delete("/{version_id}", response_model=None)
|
|
def delete_route(
|
|
version_id: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["temporal-products"]),
|
|
) -> None:
|
|
with SessionFuture() as db:
|
|
id_ = db.execute(
|
|
select(ProductVersion.product_id).where(ProductVersion.id == version_id).group_by(ProductVersion.product_id)
|
|
).scalar_one()
|
|
valid_from, valid_till = db.execute(
|
|
select(ProductVersion.valid_from, ProductVersion.valid_till).where(ProductVersion.id == version_id)
|
|
).one()
|
|
day = func.date_trunc(
|
|
"day", Voucher.date + timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES - settings.TIMEZONE_OFFSET_MINUTES)
|
|
).label("day")
|
|
query = select(count(Inventory.id)).join(Inventory.kot).join(Kot.voucher).where(Inventory.product_id == id_)
|
|
if valid_from is not None:
|
|
query = query.where(day >= valid_from)
|
|
if valid_till is not None:
|
|
query = query.where(day <= valid_till)
|
|
invs = db.execute(query).scalar_one()
|
|
if invs > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="The cannot delete this product as it was billed",
|
|
)
|
|
|
|
db.execute(
|
|
delete(ProductVersion).where(ProductVersion.id == version_id).execution_options(synchronize_session=False)
|
|
)
|
|
db.execute(
|
|
delete(Product)
|
|
.where(~Product.id.in_(select(distinct(ProductVersion.product_id))))
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
db.commit()
|
|
return
|
|
|
|
|
|
@router.get("/list", response_model=List[List[schemas.Product]])
|
|
def show_list(user: UserToken = Security(get_user, scopes=["temporal-products"])) -> List[List[schemas.Product]]:
|
|
with SessionFuture() as db:
|
|
return product_list(db)
|
|
|
|
|
|
def product_list(db: Session) -> List[List[schemas.Product]]:
|
|
dict_: Dict[uuid.UUID, List[schemas.Product]] = {}
|
|
list_ = (
|
|
db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.join(ProductVersion.sale_category)
|
|
.order_by(MenuCategory.sort_order)
|
|
.order_by(MenuCategory.name)
|
|
.order_by(ProductVersion.sort_order)
|
|
.order_by(ProductVersion.name)
|
|
.options(
|
|
joinedload(ProductVersion.menu_category, innerjoin=True),
|
|
joinedload(ProductVersion.sale_category, innerjoin=True),
|
|
contains_eager(ProductVersion.menu_category),
|
|
contains_eager(ProductVersion.sale_category),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for item in list_:
|
|
if item.product_id not in dict_:
|
|
dict_[item.product_id] = []
|
|
dict_[item.product_id].append(product_info(item))
|
|
dict_[item.product_id] = sorted(dict_[item.product_id], key=lambda k: k.valid_from or date.min)
|
|
return list(dict_.values())
|
|
|
|
|
|
@router.get("/{version_id}", response_model=schemas.Product)
|
|
def show_id(
|
|
version_id: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> schemas.Product:
|
|
with SessionFuture() as db:
|
|
item = db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.join(ProductVersion.sale_category)
|
|
.join(SaleCategory.tax)
|
|
.where(ProductVersion.id == version_id)
|
|
.order_by(ProductVersion.valid_till)
|
|
.options(
|
|
joinedload(ProductVersion.menu_category, innerjoin=True),
|
|
joinedload(ProductVersion.sale_category, innerjoin=True),
|
|
joinedload(ProductVersion.sale_category, SaleCategory.tax, innerjoin=True),
|
|
contains_eager(ProductVersion.sale_category),
|
|
contains_eager(ProductVersion.sale_category, SaleCategory.tax),
|
|
)
|
|
).scalar_one()
|
|
return product_info(item)
|
|
|
|
|
|
def product_info(item: ProductVersion) -> schemas.Product:
|
|
return schemas.Product(
|
|
id=item.product_id,
|
|
versionId=item.id,
|
|
name=item.name,
|
|
units=item.units,
|
|
menuCategory=schemas.MenuCategoryLink(id=item.menu_category_id, name=item.menu_category.name, products=[]),
|
|
saleCategory=schemas.SaleCategoryLink(
|
|
id=item.sale_category_id,
|
|
name=item.sale_category.name,
|
|
),
|
|
price=item.price,
|
|
hasHappyHour=item.has_happy_hour,
|
|
isNotAvailable=item.is_not_available,
|
|
quantity=item.quantity,
|
|
isActive=True,
|
|
sortOrder=item.sort_order,
|
|
validFrom=item.valid_from,
|
|
validTill=item.valid_till,
|
|
)
|