barker/barker/barker/routers/menu_category.py

import uuid

from datetime import date
from operator import and_, or_
from typing import List

import barker.schemas.menu_category as schemas

from fastapi import APIRouter, Depends, HTTPException, Security, status
from sqlalchemy import distinct, select, update
from sqlalchemy.exc import SQLAlchemyError

from ..core.security import get_current_active_user as get_user
from ..db.session import SessionFuture
from ..models.menu_category import MenuCategory
from ..models.product_version import ProductVersion
from ..schemas.user_token import UserToken
from . import effective_date


router = APIRouter()


@router.post("/list", response_model=List[schemas.MenuCategory])
def sort_order(
    data: List[schemas.MenuCategory],
    user: UserToken = Security(get_user, scopes=["sections"]),
) -> List[schemas.MenuCategory]:
    try:
        with SessionFuture() as db:
            for index, item in enumerate(data):
                db.execute(update(MenuCategory).where(MenuCategory.id == item.id_).values(sort_order=index))
            db.commit()
            query = select(MenuCategory).order_by(MenuCategory.sort_order).order_by(MenuCategory.name)
            return [menu_category_info(item) for item in db.execute(query).scalars().all()]
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


@router.post("", response_model=schemas.MenuCategory)
def save(
    data: schemas.MenuCategoryIn,
    user: UserToken = Security(get_user, scopes=["products"]),
) -> schemas.MenuCategory:
    try:
        with SessionFuture() as db:
            item = MenuCategory(
                name=data.name,
                is_active=data.is_active,
                sort_order=0,
            )
            db.add(item)
            db.commit()
            return menu_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.MenuCategory)
def update_route(
    id_: uuid.UUID,
    data: schemas.MenuCategoryIn,
    user: UserToken = Security(get_user, scopes=["products"]),
) -> schemas.MenuCategory:
    try:
        with SessionFuture() as db:
            item: MenuCategory = db.execute(select(MenuCategory).where(MenuCategory.id == id_)).scalar_one()
            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.",
                )
            item.name = data.name
            item.is_active = data.is_active
            db.commit()
            return menu_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.MenuCategoryBlank)
def delete_route(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["products"]),
) -> schemas.MenuCategoryBlank:
    with SessionFuture() as db:
        item: MenuCategory = db.execute(select(MenuCategory).where(MenuCategory.id == id_)).scalar_one()
        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.",
            )
        else:
            raise HTTPException(
                status_code=status.HTTP_501_NOT_IMPLEMENTED,
                detail="Menu Category deletion not implemented",
            )


@router.get("", response_model=schemas.MenuCategoryBlank)
def show_blank(
    user: UserToken = Security(get_user, scopes=["products"]),
) -> schemas.MenuCategoryBlank:
    return menu_category_blank()


@router.get("/list", response_model=List[schemas.MenuCategory])
def show_list(
    p: bool = False,
    date_: date = Depends(effective_date),
    user: UserToken = Depends(get_user),
) -> List[schemas.MenuCategory]:
    if p:
        sq = select(distinct(ProductVersion.menu_category_id)).where(
            and_(
                or_(
                    ProductVersion.valid_from == None,  # noqa: E711
                    ProductVersion.valid_from <= date_,
                ),
                or_(
                    ProductVersion.valid_till == None,  # noqa: E711
                    ProductVersion.valid_till >= date_,
                ),
            )
        )
        query = (
            select(MenuCategory)
            .where(MenuCategory.id.in_(sq), MenuCategory.is_active == True)  # noqa: E712
            .order_by(MenuCategory.sort_order)
            .order_by(MenuCategory.name)
        )
    else:
        query = select(MenuCategory).order_by(MenuCategory.sort_order).order_by(MenuCategory.name)
    with SessionFuture() as db:
        return [menu_category_info(item) for item in db.execute(query).scalars().all()]


@router.get("/{id_}", response_model=schemas.MenuCategory)
def show_id(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["products"]),
) -> schemas.MenuCategory:
    with SessionFuture() as db:
        item: MenuCategory = db.execute(select(MenuCategory).where(MenuCategory.id == id_)).scalar_one()
        return menu_category_info(item)


def menu_category_info(item: MenuCategory) -> schemas.MenuCategory:
    return schemas.MenuCategory(
        id=item.id,
        name=item.name,
        isActive=item.is_active,
        isFixture=item.is_fixture,
        sortOrder=item.sort_order,
    )


def menu_category_blank() -> schemas.MenuCategoryBlank:
    return schemas.MenuCategoryBlank(
        name="",
        isActive=True,
        isFixture=False,
        sortOrder=0,
    )