140 lines
5.0 KiB
Python
140 lines
5.0 KiB
Python
import uuid
|
|
|
|
from typing import List, Optional
|
|
|
|
import barker.schemas.section_printer as schemas
|
|
|
|
from fastapi import APIRouter, HTTPException, Security, status
|
|
from sqlalchemy import and_, delete, or_, select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.menu_category import MenuCategory
|
|
from ..models.section_printer import SectionPrinter
|
|
from ..schemas.user_token import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/{id_}", response_model=List[schemas.SectionPrinter])
|
|
def save(
|
|
id_: uuid.UUID,
|
|
data: List[schemas.SectionPrinter],
|
|
user: UserToken = Security(get_user, scopes=["section-printers"]),
|
|
) -> List[schemas.SectionPrinter]:
|
|
try:
|
|
with SessionFuture() as db:
|
|
current = []
|
|
default_id = uuid.uuid4()
|
|
for mcs in data:
|
|
if mcs.menu_category is None and mcs.printer is None:
|
|
raise ValueError("Please choose a default printer")
|
|
if mcs.printer is None:
|
|
continue
|
|
stmt = (
|
|
pg_insert(SectionPrinter)
|
|
.values(
|
|
id=default_id if mcs.menu_category is None else uuid.uuid4(),
|
|
menu_category_id=mcs.menu_category.id_ if mcs.menu_category is not None else None,
|
|
section_id=id_,
|
|
printer_id=mcs.printer.id_,
|
|
copies=mcs.copies,
|
|
)
|
|
.on_conflict_do_update(
|
|
index_elements=["menu_category_id", "section_id"],
|
|
set_=dict(printer_id=mcs.printer.id_, copies=mcs.copies),
|
|
)
|
|
)
|
|
db.execute(stmt)
|
|
current.append(None if mcs.menu_category is None else mcs.menu_category.id_)
|
|
db.execute(
|
|
delete(SectionPrinter).where(
|
|
or_(
|
|
and_(
|
|
SectionPrinter.section_id == id_,
|
|
SectionPrinter.menu_category_id == None, # noqa: E711
|
|
SectionPrinter.id != default_id,
|
|
),
|
|
and_(
|
|
SectionPrinter.section_id == id_,
|
|
SectionPrinter.menu_category_id != None, # noqa: E711
|
|
~SectionPrinter.menu_category_id.in_([x for x in current if x is not None]),
|
|
),
|
|
)
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
return report(id_, db)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.delete("/{id_}", response_model=List[schemas.SectionPrinter])
|
|
def delete_route(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["section-printers"]),
|
|
) -> List[schemas.SectionPrinter]:
|
|
try:
|
|
with SessionFuture() as db:
|
|
db.execute(delete(SectionPrinter).where(SectionPrinter.section_id == id_))
|
|
db.commit()
|
|
return report(id_, db)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[schemas.SectionPrinter])
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["section-printers"]),
|
|
) -> List[schemas.SectionPrinter]:
|
|
with SessionFuture() as db:
|
|
return report(None, db)
|
|
|
|
|
|
@router.get("/{id_}", response_model=List[schemas.SectionPrinter])
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
user: UserToken = Security(get_user, scopes=["section-printers"]),
|
|
) -> List[schemas.SectionPrinter]:
|
|
with SessionFuture() as db:
|
|
return report(id_, db)
|
|
|
|
|
|
def report(section_id: Optional[uuid.UUID], db: Session) -> List[schemas.SectionPrinter]:
|
|
menu_categories = db.execute(
|
|
select(MenuCategory.id, MenuCategory.name)
|
|
.where(MenuCategory.is_active == True) # noqa: E712
|
|
.order_by(MenuCategory.sort_order)
|
|
).all()
|
|
list_ = []
|
|
for mc_id, mc_name in [(None, None)] + menu_categories:
|
|
section_printer = (
|
|
db.execute(
|
|
select(SectionPrinter).where(
|
|
SectionPrinter.section_id == section_id,
|
|
SectionPrinter.menu_category_id == mc_id,
|
|
)
|
|
)
|
|
.scalars()
|
|
.one_or_none()
|
|
)
|
|
list_.append(
|
|
schemas.SectionPrinter(
|
|
menuCategory=None if mc_id is None else schemas.MenuCategoryLink(id=mc_id, name=mc_name, products=[]),
|
|
printer=None if section_printer is None else schemas.PrinterLink(id=section_printer.printer_id),
|
|
copies=0 if section_printer is None else section_printer.copies,
|
|
)
|
|
)
|
|
return list_
|