136 lines
4.9 KiB
Python
136 lines
4.9 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.sale_category import SaleCategory
|
|
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 scs in data:
|
|
if scs.sale_category is None and scs.printer is None:
|
|
raise ValueError("Please choose a default printer")
|
|
if scs.printer is None:
|
|
continue
|
|
stmt = (
|
|
pg_insert(SectionPrinter)
|
|
.values(
|
|
id=default_id if scs.sale_category is None else uuid.uuid4(),
|
|
sale_category_id=scs.sale_category.id_ if scs.sale_category is not None else None,
|
|
section_id=id_,
|
|
printer_id=scs.printer.id_,
|
|
copies=scs.copies,
|
|
)
|
|
.on_conflict_do_update(
|
|
index_elements=["sale_category_id", "section_id"],
|
|
set_=dict(printer_id=scs.printer.id_, copies=scs.copies),
|
|
)
|
|
)
|
|
db.execute(stmt)
|
|
current.append(None if scs.sale_category is None else scs.sale_category.id_)
|
|
db.execute(
|
|
delete(SectionPrinter).where(
|
|
or_(
|
|
and_(
|
|
SectionPrinter.section_id == id_,
|
|
SectionPrinter.sale_category_id == None, # noqa: E711
|
|
SectionPrinter.id != default_id,
|
|
),
|
|
and_(
|
|
SectionPrinter.section_id == id_,
|
|
SectionPrinter.sale_category_id != None, # noqa: E711
|
|
~SectionPrinter.sale_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]:
|
|
sale_categories = db.execute(select(SaleCategory.id, SaleCategory.name).order_by(SaleCategory.name)).all()
|
|
list_ = []
|
|
for sc_id, sc_name in [(None, None)] + sale_categories:
|
|
section_printer = (
|
|
db.execute(
|
|
select(SectionPrinter).where(
|
|
SectionPrinter.section_id == section_id,
|
|
SectionPrinter.sale_category_id == sc_id,
|
|
)
|
|
)
|
|
.scalars()
|
|
.one_or_none()
|
|
)
|
|
list_.append(
|
|
schemas.SectionPrinter(
|
|
saleCategory=None if sc_id is None else schemas.SaleCategoryLink(id=sc_id, name=sc_name),
|
|
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_
|