barker/barker/barker/routers/section.py

import uuid

from typing import List

import barker.schemas.section as schemas

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

from ..core.security import get_current_active_user as get_user
from ..db.session import SessionFuture
from ..models.section import Section
from ..models.section_printer import SectionPrinter
from ..schemas.user_token import UserToken


router = APIRouter()


@router.post("", response_model=schemas.Section)
def save(
    data: schemas.SectionIn,
    user: UserToken = Security(get_user, scopes=["sections"]),
) -> schemas.Section:
    try:
        with SessionFuture() as db:
            item = Section(name=data.name)
            db.add(item)
            db.commit()
            return section_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.Section)
def update_route(
    id_: uuid.UUID,
    data: schemas.SectionIn,
    user: UserToken = Security(get_user, scopes=["sections"]),
) -> schemas.Section:
    try:
        with SessionFuture() as db:
            item: Section = db.execute(select(Section).where(Section.id == id_)).scalar_one()
            if item.is_fixture and "temporal-products" not in user.permissions:
                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
            db.commit()
            return section_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.SectionBlank)
def delete_route(
    id_: uuid.UUID,
    user: UserToken = Security(get_user, scopes=["sections"]),
) -> schemas.SectionBlank:
    try:
        with SessionFuture() as db:
            item: Section = db.execute(select(Section).where(Section.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.",
                )
            db.execute(
                delete(SectionPrinter)
                .where(SectionPrinter.section_id == id_)
                .execution_options(synchronize_session=False)
            )
            db.execute(delete(Section).where(Section.id == id_).execution_options(synchronize_session=False))
            db.commit()
            return section_blank()
    except SQLAlchemyError as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )


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


@router.get("/list", response_model=List[schemas.Section])
def show_list(user: UserToken = Depends(get_user)):
    with SessionFuture() as db:
        return [section_info(item) for item in db.execute(select(Section).order_by(Section.name)).scalars().all()]


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


def section_info(item: Section) -> schemas.Section:
    return schemas.Section(id=item.id, name=item.name)


def section_blank() -> schemas.SectionBlank:
    return schemas.SectionBlank(name="")