124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
"""Schedule steps nested under termsheets. step_no auto-numbers (MAX+1)."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.engine import get_db
|
|
from app.models.procurement import Termsheet, TermsheetScheduleStep
|
|
from app.schemas.procurement import ScheduleStepCreate, ScheduleStepUpdate
|
|
from app.services.expand_service import schedule_step_to_dict
|
|
from app.services.quote_service import next_step_no
|
|
from app.utils.errors import ConflictError, NotFoundError
|
|
from app.utils.pagination import paginate
|
|
|
|
router = APIRouter(
|
|
prefix="/quotes/{quote_id}/termsheets/{termsheet_id}/schedule-steps",
|
|
tags=["schedule-steps"],
|
|
)
|
|
|
|
|
|
async def _termsheet_or_404(db: AsyncSession, quote_id, termsheet_id) -> Termsheet:
|
|
ts = await db.scalar(
|
|
select(Termsheet).where(Termsheet.id == termsheet_id, Termsheet.quote_id == quote_id)
|
|
)
|
|
if ts is None:
|
|
raise NotFoundError("ترمشیت یافت نشد")
|
|
return ts
|
|
|
|
|
|
async def _step_or_404(db: AsyncSession, termsheet_id, step_id) -> TermsheetScheduleStep:
|
|
step = await db.scalar(
|
|
select(TermsheetScheduleStep).where(
|
|
TermsheetScheduleStep.id == step_id, TermsheetScheduleStep.termsheet_id == termsheet_id
|
|
)
|
|
)
|
|
if step is None:
|
|
raise NotFoundError("مرحله پرداخت یافت نشد")
|
|
return step
|
|
|
|
|
|
@router.get("")
|
|
async def list_schedule_steps(
|
|
quote_id,
|
|
termsheet_id,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
await _termsheet_or_404(db, quote_id, termsheet_id)
|
|
stmt = (
|
|
select(TermsheetScheduleStep)
|
|
.where(TermsheetScheduleStep.termsheet_id == termsheet_id)
|
|
.order_by(TermsheetScheduleStep.step_no.asc())
|
|
)
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [schedule_step_to_dict(s) for s in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_schedule_step(
|
|
quote_id, termsheet_id, payload: ScheduleStepCreate, db: AsyncSession = Depends(get_db)
|
|
):
|
|
ts = await _termsheet_or_404(db, quote_id, termsheet_id)
|
|
step_no = payload.step_no
|
|
if step_no is None:
|
|
step_no = await next_step_no(db, ts.id)
|
|
duplicate = await db.scalar(
|
|
select(TermsheetScheduleStep.id).where(
|
|
TermsheetScheduleStep.termsheet_id == ts.id, TermsheetScheduleStep.step_no == step_no
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("شماره مرحله تکراری است", field="step_no")
|
|
step = TermsheetScheduleStep(
|
|
termsheet_id=ts.id,
|
|
step_no=step_no,
|
|
due_date=payload.due_date,
|
|
amount=payload.amount,
|
|
note=payload.note,
|
|
)
|
|
db.add(step)
|
|
await db.flush()
|
|
return {"data": schedule_step_to_dict(step)}
|
|
|
|
|
|
@router.patch("/{step_id}")
|
|
async def update_schedule_step(
|
|
quote_id, termsheet_id, step_id, payload: ScheduleStepUpdate, db: AsyncSession = Depends(get_db)
|
|
):
|
|
step = await _step_or_404(db, termsheet_id, step_id)
|
|
if payload.step_no is not None and payload.step_no != step.step_no:
|
|
duplicate = await db.scalar(
|
|
select(TermsheetScheduleStep.id).where(
|
|
TermsheetScheduleStep.termsheet_id == termsheet_id,
|
|
TermsheetScheduleStep.step_no == payload.step_no,
|
|
TermsheetScheduleStep.id != step_id,
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("شماره مرحله تکراری است", field="step_no")
|
|
step.step_no = payload.step_no
|
|
if payload.due_date is not None:
|
|
step.due_date = payload.due_date
|
|
if payload.amount is not None:
|
|
step.amount = payload.amount
|
|
if payload.note is not None:
|
|
step.note = payload.note
|
|
await db.flush()
|
|
return {"data": schedule_step_to_dict(step)}
|
|
|
|
|
|
@router.delete("/{step_id}")
|
|
async def delete_schedule_step(quote_id, termsheet_id, step_id, db: AsyncSession = Depends(get_db)):
|
|
step = await _step_or_404(db, termsheet_id, step_id)
|
|
count = await db.scalar(
|
|
select(func.count()).select_from(TermsheetScheduleStep).where(
|
|
TermsheetScheduleStep.termsheet_id == termsheet_id
|
|
)
|
|
)
|
|
if count <= 1:
|
|
raise ConflictError("آخرین مرحله پرداخت قابل حذف نیست")
|
|
await db.delete(step)
|
|
await db.flush()
|
|
return {"data": {"deleted": True}}
|