129 lines
5.1 KiB
Python
129 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Annotated
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.constants import ERROR_CODE_OK
|
|
from app.core.config import get_settings
|
|
from app.core.time import format_now
|
|
from app.db.session import get_db
|
|
from app.schemas.approval import (
|
|
ApprovalDecisionRequest,
|
|
ApprovalDetailData,
|
|
ApprovalListData,
|
|
CreateApprovalData,
|
|
CreateApprovalRequest,
|
|
)
|
|
from app.schemas.common import ApiResponse
|
|
from app.services.approval_service import ApprovalConflictError, ApprovalNotFoundError, ApprovalService
|
|
|
|
router = APIRouter(prefix="/api/demo/approval", tags=["demo-approval"])
|
|
|
|
|
|
def build_request_id(header_value: str | None) -> str:
|
|
return header_value or f"req-{uuid4().hex[:12]}"
|
|
|
|
|
|
def to_detail_data(approval) -> ApprovalDetailData:
|
|
return ApprovalDetailData(
|
|
approval_id=approval.approval_id,
|
|
task_id=approval.task_id,
|
|
approval_status=approval.approval_status,
|
|
risk_level=approval.risk_level,
|
|
approvers=json.loads(approval.approver_user_ids_json),
|
|
reason=approval.reason,
|
|
created_at=approval.created_at,
|
|
updated_at=approval.updated_at,
|
|
)
|
|
|
|
|
|
@router.post("/requests", response_model=ApiResponse[CreateApprovalData])
|
|
def create_request(
|
|
payload: CreateApprovalRequest,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[CreateApprovalData]:
|
|
request_id = build_request_id(x_request_id)
|
|
approval = ApprovalService(db, get_settings().default_timezone).create_request(payload)
|
|
return ApiResponse[CreateApprovalData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="approval created",
|
|
data=CreateApprovalData(approval_id=approval.approval_id, approval_status=approval.approval_status),
|
|
timestamp=format_now(get_settings().default_timezone),
|
|
)
|
|
|
|
|
|
@router.get("/requests/{approval_id}", response_model=ApiResponse[ApprovalDetailData])
|
|
def get_request(
|
|
approval_id: str,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[ApprovalDetailData]:
|
|
request_id = build_request_id(x_request_id)
|
|
service = ApprovalService(db, get_settings().default_timezone)
|
|
try:
|
|
approval = service.get_request(approval_id)
|
|
except ApprovalNotFoundError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail={"code": exc.code, "message": "approval not found"}) from exc
|
|
return ApiResponse[ApprovalDetailData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="success",
|
|
data=to_detail_data(approval),
|
|
timestamp=format_now(get_settings().default_timezone),
|
|
)
|
|
|
|
|
|
@router.post("/requests/{approval_id}/decision", response_model=ApiResponse[ApprovalDetailData])
|
|
def decide_request(
|
|
approval_id: str,
|
|
payload: ApprovalDecisionRequest,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[ApprovalDetailData]:
|
|
request_id = build_request_id(x_request_id)
|
|
service = ApprovalService(db, get_settings().default_timezone)
|
|
try:
|
|
approval = service.decide(approval_id, payload)
|
|
except ApprovalNotFoundError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail={"code": exc.code, "message": "approval not found"}) from exc
|
|
except ApprovalConflictError as exc:
|
|
message = exc.args[0] if exc.args else "approval conflict"
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail={"code": exc.code, "message": message}) from exc
|
|
return ApiResponse[ApprovalDetailData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="success",
|
|
data=to_detail_data(approval),
|
|
timestamp=format_now(get_settings().default_timezone),
|
|
)
|
|
|
|
|
|
@router.get("/requests", response_model=ApiResponse[ApprovalListData])
|
|
def list_pending_requests(
|
|
approval_status: str = Query(default="PENDING"),
|
|
approver_user_id: str | None = Query(default=None),
|
|
db: Annotated[Session, Depends(get_db)] = None,
|
|
x_request_id: Annotated[str | None, Header(alias="X-Request-Id")] = None,
|
|
) -> ApiResponse[ApprovalListData]:
|
|
request_id = build_request_id(x_request_id)
|
|
if approval_status != "PENDING":
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail={"code": "INVALID_PARAM", "message": "当前仅支持查询 PENDING 审批单"})
|
|
approvals = ApprovalService(db, get_settings().default_timezone).list_pending(approver_user_id)
|
|
return ApiResponse[ApprovalListData](
|
|
request_id=request_id,
|
|
success=True,
|
|
code=ERROR_CODE_OK,
|
|
message="success",
|
|
data=ApprovalListData(approvals=[to_detail_data(item) for item in approvals]),
|
|
timestamp=format_now(get_settings().default_timezone),
|
|
)
|