569 lines
21 KiB
Python
569 lines
21 KiB
Python
import uuid
|
|
|
|
from datetime import date, timedelta
|
|
from typing import List, Optional
|
|
|
|
import barker.schemas.product as schemas
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import and_, delete, insert, or_, select, update
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session, contains_eager, joinedload
|
|
from sqlalchemy.sql.functions import count
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.menu_category import MenuCategory
|
|
from ..models.modifier_categories_products import modifier_categories_products
|
|
from ..models.modifier_category import ModifierCategory
|
|
from ..models.product import Product
|
|
from ..models.product_version import ProductVersion
|
|
from ..models.sale_category import SaleCategory
|
|
from ..schemas.user_token import UserToken
|
|
from . import effective_date
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/list", response_model=List[schemas.Product])
|
|
def sort_order(
|
|
data: List[schemas.Product],
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
):
|
|
try:
|
|
with SessionFuture() as db:
|
|
indexes = {}
|
|
for item in data:
|
|
if item.menu_category.id_ in indexes:
|
|
indexes[item.menu_category.id_] += 1
|
|
else:
|
|
indexes[item.menu_category.id_] = 0
|
|
db.execute(
|
|
update(ProductVersion)
|
|
.where(
|
|
and_(
|
|
ProductVersion.product_id == item.id_,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
.values(sort_order=indexes[item.menu_category.id_])
|
|
)
|
|
db.commit()
|
|
return product_list(date_, db)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.post("", response_model=schemas.Product)
|
|
def save(
|
|
data: schemas.ProductIn,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item = Product()
|
|
db.add(item)
|
|
db.flush()
|
|
product_version = ProductVersion(
|
|
product_id=item.id,
|
|
name=data.name,
|
|
units=data.units,
|
|
menu_category_id=data.menu_category.id_,
|
|
sale_category_id=data.sale_category.id_,
|
|
price=data.price,
|
|
has_happy_hour=data.has_happy_hour,
|
|
is_not_available=data.is_not_available,
|
|
quantity=data.quantity,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
)
|
|
db.add(product_version)
|
|
add_modifiers(item.id, product_version.menu_category_id, date_, db)
|
|
db.commit()
|
|
return
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
def add_modifiers(product_id: uuid.UUID, menu_category_id: uuid.UUID, date_: date, db: Session):
|
|
products_in_category = db.execute(
|
|
select(count(ProductVersion.id)).where(
|
|
and_(
|
|
ProductVersion.menu_category_id == menu_category_id,
|
|
ProductVersion.product_id != product_id,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
).scalar()
|
|
categories = db.execute(
|
|
select(count(ProductVersion.id), ModifierCategory.id)
|
|
.join(Product.versions)
|
|
.join(modifier_categories_products, ProductVersion.id == modifier_categories_products.c.product_id)
|
|
.join(ModifierCategory, ModifierCategory.id == modifier_categories_products.c.modifier_category_id)
|
|
.where(
|
|
and_(
|
|
ProductVersion.menu_category_id == menu_category_id,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
.group_by(ModifierCategory.id)
|
|
).all()
|
|
for c, mc in categories:
|
|
if c == products_in_category:
|
|
db.execute(insert(modifier_categories_products).values(product_id=product_id, modifier_category_id=mc))
|
|
|
|
|
|
@router.put("/{version_id}", response_model=schemas.Product)
|
|
def update_route(
|
|
version_id: uuid.UUID,
|
|
data: schemas.ProductIn,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
try:
|
|
with SessionFuture() as db:
|
|
old: ProductVersion = db.execute(select(ProductVersion).where(ProductVersion.id == version_id)).scalar_one()
|
|
id_ = old.product_id
|
|
latest: ProductVersion = db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.where(
|
|
and_(
|
|
ProductVersion.product_id == id_,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
).scalar_one()
|
|
if version_id != latest.id and "temporal-products" not in user.permissions:
|
|
# This should not happen as only someone with this permission should reach here
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Permission error, you cannot edit this product version.",
|
|
)
|
|
|
|
if version_id != latest.id:
|
|
# Update the old product update by temporal product editor
|
|
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
|
|
db.commit()
|
|
return
|
|
|
|
if latest.valid_till is not None:
|
|
# Allow adding a product here splitting the valid from and to, but not implemented right now
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Product has been invalidated",
|
|
)
|
|
|
|
if latest.valid_from == date_:
|
|
# Update the product as it is valid from the the same
|
|
latest.name = data.name
|
|
latest.units = data.units
|
|
latest.menu_category_id = data.menu_category.id_
|
|
latest.sale_category_id = data.sale_category.id_
|
|
latest.price = data.price
|
|
latest.has_happy_hour = data.has_happy_hour
|
|
latest.is_not_available = data.is_not_available
|
|
latest.quantity = data.quantity
|
|
db.commit()
|
|
return
|
|
|
|
# Create a new product version
|
|
latest.valid_till = date_ - timedelta(days=1)
|
|
product_version = ProductVersion(
|
|
product_id=id_,
|
|
name=data.name,
|
|
units=data.units,
|
|
menu_category_id=data.menu_category.id_,
|
|
sale_category_id=data.sale_category.id_,
|
|
price=data.price,
|
|
has_happy_hour=data.has_happy_hour,
|
|
is_not_available=data.is_not_available,
|
|
quantity=data.quantity,
|
|
valid_from=date_,
|
|
valid_till=None,
|
|
sort_order=latest.sort_order,
|
|
)
|
|
db.add(product_version)
|
|
db.commit()
|
|
return
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.delete("/{version_id}")
|
|
def delete_route(
|
|
version_id: uuid.UUID,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> None:
|
|
with SessionFuture() as db:
|
|
old: ProductVersion = db.execute(select(ProductVersion).where(ProductVersion.id == version_id)).scalar_one()
|
|
id_ = old.product_id
|
|
latest: ProductVersion = db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.where(
|
|
and_(
|
|
ProductVersion.product_id == id_,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
).scalar_one()
|
|
if version_id != latest.id and "temporal-products" not in user.permissions:
|
|
# This should not happen as only someone with this permission should reach here
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Permission error, you cannot delete this product.",
|
|
)
|
|
|
|
if version_id != latest.id:
|
|
# Delete old product, but make sure that no gaps remain
|
|
if old.valid_from is not None:
|
|
# Set the previous version valid till to item's valid till
|
|
db.execute(
|
|
delete(ProductVersion)
|
|
.where(ProductVersion.id == version_id)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
id_to_update = (
|
|
select(ProductVersion.id)
|
|
.where(
|
|
ProductVersion.product_id == id_,
|
|
ProductVersion.valid_till == old.valid_from - timedelta(days=1),
|
|
)
|
|
.scalar_subquery()
|
|
)
|
|
db.execute(
|
|
update(ProductVersion)
|
|
.where(ProductVersion.id == id_to_update)
|
|
.values(valid_till=old.valid_till)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
else:
|
|
# Set the next version valid from to item's valid from which is None
|
|
db.execute(
|
|
delete(ProductVersion)
|
|
.where(ProductVersion.id == version_id)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
id_to_update = (
|
|
select(ProductVersion.id)
|
|
.where(
|
|
ProductVersion.product_id == id_,
|
|
ProductVersion.valid_from == old.valid_till + timedelta(days=1),
|
|
)
|
|
.scalar_subquery()
|
|
)
|
|
db.execute(
|
|
update(ProductVersion)
|
|
.where(ProductVersion.id == id_to_update)
|
|
.values(valid_till=old.valid_from)
|
|
.execution_options(synchronize_session=False)
|
|
)
|
|
db.commit()
|
|
return
|
|
if latest.valid_from == date_:
|
|
db.delete(latest)
|
|
else:
|
|
latest.valid_till = date_ - timedelta(days=1)
|
|
db.commit()
|
|
return
|
|
|
|
|
|
@router.get("", response_model=List[schemas.ProductBlank])
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> List[schemas.ProductBlank]:
|
|
return [
|
|
schemas.ProductBlank(
|
|
name="",
|
|
units="",
|
|
price=0,
|
|
hasHappyHour=False,
|
|
isNotAvailable=False,
|
|
isActive=True,
|
|
sortOrder=0,
|
|
)
|
|
]
|
|
|
|
|
|
@router.get("/list", response_model=List[schemas.Product])
|
|
def show_list(date_: date = Depends(effective_date), user: UserToken = Depends(get_user)) -> List[schemas.Product]:
|
|
with SessionFuture() as db:
|
|
return product_list(date_, db)
|
|
|
|
|
|
def product_list(date_: date, db: Session) -> List[schemas.Product]:
|
|
return [
|
|
product_info(item)
|
|
for item in db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.join(ProductVersion.sale_category)
|
|
.where(
|
|
and_(
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
.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()
|
|
]
|
|
|
|
|
|
@router.get("/query")
|
|
async def show_term(
|
|
mc: Optional[uuid.UUID] = None,
|
|
sc: Optional[uuid.UUID] = None,
|
|
date_: date = Depends(effective_date),
|
|
current_user: UserToken = Depends(get_user),
|
|
):
|
|
list_ = []
|
|
query = select(ProductVersion)
|
|
if sc is not None:
|
|
query = query.join(ProductVersion.menu_category)
|
|
if mc is not None:
|
|
query = query.join(ProductVersion.sale_category).join(SaleCategory.tax)
|
|
query = query.where(
|
|
and_(
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
if mc is not None:
|
|
query = query.where(ProductVersion.menu_category_id == mc).order_by(
|
|
ProductVersion.sort_order, ProductVersion.name
|
|
)
|
|
if sc is not None:
|
|
query = query.where(ProductVersion.sale_category_id == sc).order_by(
|
|
MenuCategory.sort_order, ProductVersion.sort_order, ProductVersion.name
|
|
)
|
|
|
|
if mc is not None:
|
|
query = query.options(
|
|
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),
|
|
)
|
|
|
|
if sc is not None:
|
|
query = query.options(
|
|
joinedload(ProductVersion.menu_category, innerjoin=True),
|
|
contains_eager(ProductVersion.menu_category),
|
|
)
|
|
|
|
with SessionFuture() as db:
|
|
for item in db.execute(query).scalars().all():
|
|
if sc is not None:
|
|
list_.append(
|
|
{
|
|
"id": item.product_id,
|
|
"name": item.full_name,
|
|
"menuCategory": {
|
|
"id": item.menu_category_id,
|
|
"name": item.menu_category.name,
|
|
},
|
|
"price": item.price,
|
|
}
|
|
)
|
|
if mc is not None:
|
|
list_.append(query_product_info(item, False))
|
|
if item.has_happy_hour:
|
|
list_.append(query_product_info(item, True))
|
|
return list_
|
|
|
|
|
|
def product_list_of_sale_category(date_: date, db: Session) -> List[schemas.Product]:
|
|
return [
|
|
product_info(item)
|
|
for item in db.execute(
|
|
select(ProductVersion)
|
|
.join(ProductVersion.menu_category)
|
|
.join(ProductVersion.sale_category)
|
|
.where(
|
|
and_(
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
)
|
|
.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()
|
|
]
|
|
|
|
|
|
@router.get("/{id_}", response_model=List[schemas.Product])
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
date_: date = Depends(effective_date),
|
|
user: UserToken = Security(get_user, scopes=["products"]),
|
|
) -> List[schemas.Product]:
|
|
query = (
|
|
select(ProductVersion)
|
|
.join(ProductVersion.sale_category)
|
|
.join(SaleCategory.tax)
|
|
.where(ProductVersion.product_id == id_)
|
|
)
|
|
if "temporal-products" not in user.permissions:
|
|
query = query.where(
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= date_,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= date_,
|
|
),
|
|
)
|
|
with SessionFuture() as db:
|
|
items = [
|
|
product_info(item)
|
|
for item in db.execute(
|
|
query.order_by(ProductVersion.valid_till).options(
|
|
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),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
]
|
|
return items
|
|
|
|
|
|
def query_product_info(item: ProductVersion, happy_hour: bool):
|
|
return {
|
|
"id": item.product_id,
|
|
"name": ("H H " if happy_hour else "") + item.full_name,
|
|
"saleCategory": {
|
|
"id": item.sale_category_id,
|
|
"name": item.sale_category.name,
|
|
},
|
|
"tax": {
|
|
"id": item.sale_category.tax_id,
|
|
"name": item.sale_category.tax.name,
|
|
"rate": item.sale_category.tax.rate,
|
|
},
|
|
"price": item.price,
|
|
"hasHappyHour": happy_hour,
|
|
"isNotAvailable": item.is_not_available,
|
|
"sortOrder": item.sort_order,
|
|
}
|
|
|
|
|
|
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,
|
|
)
|