127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import barker.schemas.regime as schemas
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.sql.functions import count
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.bill import Bill
|
|
from ..models.regime import Regime
|
|
from ..models.tax import Tax
|
|
from ..schemas.user_token import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("", response_model=schemas.Regime)
|
|
def save(
|
|
data: schemas.RegimeIn,
|
|
user: UserToken = Security(get_user, scopes=["regimes"]),
|
|
) -> schemas.Regime:
|
|
try:
|
|
with SessionFuture() as db:
|
|
regime_id = db.execute(select(func.coalesce(func.max(Regime.id), 0) + 1)).scalar_one()
|
|
item = Regime(id=regime_id, name=data.name.strip(), header=data.header, prefix=data.prefix.strip())
|
|
db.add(item)
|
|
db.commit()
|
|
return regime_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.Regime)
|
|
def update_route(
|
|
id_: int,
|
|
data: schemas.RegimeIn,
|
|
user: UserToken = Security(get_user, scopes=["regimes"]),
|
|
) -> schemas.Regime:
|
|
try:
|
|
with SessionFuture() as db:
|
|
item: Regime = db.execute(select(Regime).where(Regime.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.strip()
|
|
item.header = data.header
|
|
item.prefix = data.prefix.strip()
|
|
db.commit()
|
|
return regime_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.RegimeBlank)
|
|
def delete_route(
|
|
id_: int,
|
|
user: UserToken = Security(get_user, scopes=["regimes"]),
|
|
) -> schemas.RegimeBlank:
|
|
with SessionFuture() as db:
|
|
item: Regime = db.execute(select(Regime).where(Regime.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.",
|
|
)
|
|
if db.execute(select(count(Tax.regime_id)).where(Tax.regime_id == item.id)).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_423_LOCKED,
|
|
detail=f"{item.name} has associated Taxes and cannot be deleted",
|
|
)
|
|
if db.execute(select(count(Bill.regime_id)).where(Bill.regime_id == item.id)).scalar_one() > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_423_LOCKED,
|
|
detail=f"{item.name} has associated Bills and cannot be deleted",
|
|
)
|
|
db.delete(item)
|
|
db.commit()
|
|
return regime_blank()
|
|
|
|
|
|
@router.get("", response_model=schemas.RegimeBlank)
|
|
def show_blank(
|
|
user: UserToken = Security(get_user, scopes=["regimes"]),
|
|
) -> schemas.RegimeBlank:
|
|
return regime_blank()
|
|
|
|
|
|
@router.get("/list", response_model=list[schemas.Regime])
|
|
def show_list(user: UserToken = Depends(get_user)) -> list[schemas.Regime]:
|
|
with SessionFuture() as db:
|
|
items = db.execute(select(Regime).order_by(Regime.name)).scalars().all()
|
|
return [regime_info(item) for item in items]
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.Regime)
|
|
def show_id(
|
|
id_: int,
|
|
user: UserToken = Security(get_user, scopes=["regimes"]),
|
|
) -> schemas.Regime:
|
|
with SessionFuture() as db:
|
|
item: Regime = db.execute(select(Regime).where(Regime.id == id_)).scalar_one()
|
|
return regime_info(item)
|
|
|
|
|
|
def regime_info(item: Regime) -> schemas.Regime:
|
|
return schemas.Regime(
|
|
id=item.id,
|
|
name=item.name,
|
|
header=item.header,
|
|
prefix=item.prefix,
|
|
isFixture=item.is_fixture,
|
|
)
|
|
|
|
|
|
def regime_blank() -> schemas.RegimeBlank:
|
|
return schemas.RegimeBlank(name="", header="", prefix="", isFixture=False)
|