68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_helpdesk.backend.domain import HelpdeskDomainError, ServiceProfile
|
|
from govoplan_helpdesk.backend.schemas import ServiceProfileListResponse, ServiceProfileWriteRequest
|
|
from govoplan_helpdesk.backend.service import (
|
|
ADMIN_SCOPE,
|
|
HelpdeskConflictError,
|
|
HelpdeskStoreError,
|
|
list_service_profiles,
|
|
upsert_service_profile,
|
|
)
|
|
|
|
|
|
def create_router(registry: object | None) -> APIRouter:
|
|
router = APIRouter(prefix="/helpdesk", tags=["helpdesk"])
|
|
|
|
@router.get("/service-profiles", response_model=ServiceProfileListResponse)
|
|
def api_list_profiles(
|
|
include_inactive: bool = True,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ServiceProfileListResponse:
|
|
_require(principal)
|
|
items = list_service_profiles(session, principal, include_inactive=include_inactive)
|
|
return ServiceProfileListResponse(profiles=[item.to_dict() for item in items])
|
|
|
|
@router.put("/service-profiles/{profile_id}", response_model=dict)
|
|
def api_write_profile(
|
|
profile_id: str,
|
|
payload: ServiceProfileWriteRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> dict:
|
|
_require(principal)
|
|
if str(payload.profile.get("profile_id") or "") != profile_id:
|
|
raise HTTPException(status_code=400, detail="Helpdesk profile path and payload identifiers differ")
|
|
try:
|
|
item = upsert_service_profile(
|
|
session,
|
|
principal,
|
|
profile=ServiceProfile.from_mapping(payload.profile),
|
|
expected_revision=payload.expected_revision,
|
|
idempotency_key=payload.idempotency_key,
|
|
registry=registry,
|
|
)
|
|
session.commit()
|
|
except (HelpdeskStoreError, HelpdeskDomainError, IntegrityError) as exc:
|
|
session.rollback()
|
|
code = 409 if isinstance(exc, (HelpdeskConflictError, IntegrityError)) or "conflict" in str(exc).casefold() else 400
|
|
raise HTTPException(status_code=code, detail=str(exc)) from exc
|
|
return item.to_dict()
|
|
|
|
return router
|
|
|
|
|
|
def _require(principal: ApiPrincipal) -> None:
|
|
if not has_scope(principal, ADMIN_SCOPE):
|
|
raise HTTPException(status_code=403, detail=f"Missing scope: {ADMIN_SCOPE}")
|
|
|
|
|
|
__all__ = ["create_router"]
|