spencerm/pesco-ncr
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Use NCR business number in URLs (/ncrs/NCR-2026-0021)

Browse Source 
Detail routes, queue navigation, and notification email links now use the
business identifier instead of the database id — friendlier for end users
quoting NCR numbers. The API resolves both forms (case-insensitive number,
or legacy numeric id) so existing bookmarks and email links keep working.
Adds regression tests incl. a guard that /ncrs/export.csv isn't shadowed
by the path parameter.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ang3l12
2026-07-28 11:15:16 -06:00
parent 12383b2225
commit 8d48f774bb
7 changed files with 102 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
backend/app/routers/ncrs.py
View File

@@ -71,8 +71,17 @@ _STAGE_ORDER = [
# ── helpers ──────────────────────────────────────────────────────────────────
async def _get_ncr(db: AsyncSession, ncr_id: int) -> Ncr:
ncr = await db.get(Ncr, ncr_id)
async def _get_ncr(db: AsyncSession, ref: str | int) -> Ncr:
"""Resolve an NCR by its business identifier (e.g. NCR-2026-0021,
case-insensitive) or — for backward compatibility with older links —
by numeric database id."""
ref = str(ref).strip()
if ref.isdigit():
ncr = await db.get(Ncr, int(ref))
else:
ncr = (
await db.execute(select(Ncr).where(Ncr.ncr_number == ref.upper()))
).scalar_one_or_none()
if ncr is None:
raise HTTPException(status_code=404, detail="NCR not found.")
return ncr
@@ -453,27 +462,27 @@ async def export_ncrs_csv(
)
@router.get("/ncrs/{ncr_id}", response_model=NcrDetailOut)
@router.get("/ncrs/{ncr_ref}", response_model=NcrDetailOut)
async def get_ncr(
ncr_id: int,
ncr_ref: str,
current: CurrentUser = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> NcrDetailOut:
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
return _detail(ncr, current)
# ── stage actions ────────────────────────────────────────────────────────────
@router.post("/ncrs/{ncr_id}/initial-disposition", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/initial-disposition", response_model=NcrMutationOut)
async def initial_disposition(
ncr_id: int,
ncr_ref: str,
payload: InitialDispositionIn,
current: CurrentUser = Depends(require_roles(Role.DISPOSITION_AUTHORITY)),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Stage 2 — Initial Disposition, performed on a New Request. Routes to
Secondary Disposition (when secondary review is needed) or Operations."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
_ensure_stage(ncr, Stage.NEW_REQUEST)
assignees: list[User] = []
@@ -536,16 +545,16 @@ async def initial_disposition(
return NcrMutationOut(ncr=_detail(ncr, current), warnings=warnings)
@router.post("/ncrs/{ncr_id}/secondary-disposition", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/secondary-disposition", response_model=NcrMutationOut)
async def secondary_disposition(
ncr_id: int,
ncr_ref: str,
payload: SecondaryDispositionIn,
current: CurrentUser = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Stage 3 — Secondary Disposition. Only the assigned secondary
authorities (or an Admin) may update or release to Operations."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
_ensure_stage(ncr, Stage.SECONDARY_DISPOSITION)
if not (current.is_admin or _is_secondary_assignee(ncr, current)):
raise HTTPException(
@@ -577,14 +586,14 @@ async def secondary_disposition(
return NcrMutationOut(ncr=_detail(ncr, current), warnings=warnings)
@router.post("/ncrs/{ncr_id}/operations-complete", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/operations-complete", response_model=NcrMutationOut)
async def operations_complete(
ncr_id: int,
ncr_ref: str,
current: CurrentUser = Depends(require_roles(Role.OPERATIONS)),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Stage 4 — Operations marks rework complete; NCR moves to QC Inspection."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
_ensure_stage(ncr, Stage.OPERATIONS)
apply_field_updates(
@@ -612,16 +621,16 @@ async def operations_complete(
return NcrMutationOut(ncr=_detail(ncr, current), warnings=warnings)
@router.post("/ncrs/{ncr_id}/inspection", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/inspection", response_model=NcrMutationOut)
async def inspection(
ncr_id: int,
ncr_ref: str,
payload: InspectionIn,
current: CurrentUser = Depends(require_roles(Role.QC_INSPECTOR)),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Stage 5 — QC Inspection. QC can save repeatedly; checking QC Closed
advances the NCR to Costing."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
_ensure_stage(ncr, Stage.QC_INSPECTION)
updates = payload.model_dump(exclude_unset=True, exclude={"qc_closed"})
@@ -649,15 +658,15 @@ async def inspection(
return NcrMutationOut(ncr=_detail(ncr, current), warnings=warnings)
@router.post("/ncrs/{ncr_id}/costing", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/costing", response_model=NcrMutationOut)
async def costing(
ncr_id: int,
ncr_ref: str,
payload: CostingIn,
current: CurrentUser = Depends(require_roles(Role.COSTING)),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Stage 6 — Costing. Saving costs completes the workflow and closes the NCR."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
_ensure_stage(ncr, Stage.COSTING)
now = utcnow()
@@ -691,16 +700,16 @@ async def costing(
return NcrMutationOut(ncr=_detail(ncr, current), warnings=warnings)
@router.post("/ncrs/{ncr_id}/reopen", response_model=NcrMutationOut)
@router.post("/ncrs/{ncr_ref}/reopen", response_model=NcrMutationOut)
async def reopen(
ncr_id: int,
ncr_ref: str,
payload: ReopenIn,
current: CurrentUser = Depends(require_roles(Role.ADMIN)),
db: AsyncSession = Depends(get_db),
) -> NcrMutationOut:
"""Admin-only: reopen a closed NCR into a chosen prior stage. The reason
is required and recorded in the audit trail and transition history."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
if ncr.stage != Stage.CLOSED.value:
raise HTTPException(status_code=409, detail="Only closed NCRs can be reopened.")
if payload.to_stage == Stage.SECONDARY_DISPOSITION and not ncr.secondary_assignee_rows:
@@ -756,16 +765,16 @@ def _do_transition(
# ── attachments ──────────────────────────────────────────────────────────────
@router.post("/ncrs/{ncr_id}/attachments", response_model=list[AttachmentOut], status_code=201)
@router.post("/ncrs/{ncr_ref}/attachments", response_model=list[AttachmentOut], status_code=201)
async def upload_attachments(
ncr_id: int,
ncr_ref: str,
files: list[UploadFile],
current: CurrentUser = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> list[AttachmentOut]:
"""Photo/file attachments (multiple per request; camera capture on
tablets posts here too). Blocked once the NCR is closed."""
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
if ncr.stage == Stage.CLOSED.value:
raise HTTPException(
status_code=409, detail="This NCR is closed; attachments are locked."
@@ -818,21 +827,21 @@ async def download_attachment(
# ── audit history ────────────────────────────────────────────────────────────
@router.get("/ncrs/{ncr_id}/audit", response_model=AuditListOut)
@router.get("/ncrs/{ncr_ref}/audit", response_model=AuditListOut)
async def ncr_audit(
ncr_id: int,
ncr_ref: str,
current: CurrentUser = Depends(require_roles(Role.QC_INSPECTOR)),
db: AsyncSession = Depends(get_db),
) -> AuditListOut:
"""Audit History tab — Admin and QC roles."""
from app.models import AuditLog
await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
rows = (
(
await db.execute(
select(AuditLog)
.where(AuditLog.ncr_id == ncr_id)
.where(AuditLog.ncr_id == ncr.id)
.order_by(AuditLog.created_at.desc(), AuditLog.id.desc())
)
)
@@ -845,9 +854,9 @@ async def ncr_audit(
# ── printable PDF ────────────────────────────────────────────────────────────
@router.get("/ncrs/{ncr_id}/pdf")
@router.get("/ncrs/{ncr_ref}/pdf")
async def ncr_pdf(
ncr_id: int,
ncr_ref: str,
current: CurrentUser = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> Response:
@@ -855,7 +864,7 @@ async def ncr_pdf(
travelers and audits."""
from app.services.pdf import render_ncr_pdf
ncr = await _get_ncr(db, ncr_id)
ncr = await _get_ncr(db, ncr_ref)
pdf_bytes = await render_ncr_pdf(ncr)
return Response(
content=pdf_bytes,