130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
|
|
from govoplan_core.core.institutional import InstitutionalContextError
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_mandates.backend.manifest import READ_SCOPE, WRITE_SCOPE
|
|
from govoplan_mandates.backend.schemas import (
|
|
MandateListResponse,
|
|
MandateResolutionPayload,
|
|
MandateWriteRequest,
|
|
)
|
|
from govoplan_mandates.backend.service import (
|
|
MandateStoreError,
|
|
SqlMandateResolver,
|
|
definition_from_mapping,
|
|
get_mandate,
|
|
list_mandates,
|
|
record_mandate,
|
|
resolution_request_from_mapping,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/mandates", tags=["mandates"])
|
|
|
|
|
|
def _require_scope(principal: ApiPrincipal, scope: str) -> None:
|
|
if not has_scope(principal, scope):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Missing scope: {scope}",
|
|
)
|
|
|
|
|
|
def _error(exc: Exception) -> HTTPException:
|
|
message = str(exc)
|
|
code = status.HTTP_409_CONFLICT if "conflict" in message.lower() else status.HTTP_400_BAD_REQUEST
|
|
return HTTPException(status_code=code, detail=message)
|
|
|
|
|
|
@router.get("/definitions", response_model=MandateListResponse)
|
|
def api_list_mandates(
|
|
mandate_status: str | None = Query(default=None, alias="status"),
|
|
limit: int = Query(default=100, ge=1, le=200),
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> MandateListResponse:
|
|
_require_scope(principal, READ_SCOPE)
|
|
try:
|
|
items = list_mandates(
|
|
session,
|
|
principal,
|
|
status=mandate_status,
|
|
limit=limit,
|
|
)
|
|
except (MandateStoreError, InstitutionalContextError) as exc:
|
|
raise _error(exc) from exc
|
|
return MandateListResponse(
|
|
mandates=[item.to_dict(include_inspection=True) for item in items]
|
|
)
|
|
|
|
|
|
@router.get("/definitions/{mandate_id}", response_model=dict[str, Any])
|
|
def api_get_mandate(
|
|
mandate_id: str,
|
|
revision: str | None = None,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> dict[str, Any]:
|
|
_require_scope(principal, READ_SCOPE)
|
|
item = get_mandate(
|
|
session,
|
|
principal,
|
|
mandate_id=mandate_id,
|
|
revision=revision,
|
|
)
|
|
if item is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Mandate not found")
|
|
return item.to_dict(include_inspection=True)
|
|
|
|
|
|
@router.post(
|
|
"/definitions",
|
|
response_model=dict[str, Any],
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_record_mandate(
|
|
payload: MandateWriteRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> dict[str, Any]:
|
|
_require_scope(principal, WRITE_SCOPE)
|
|
try:
|
|
item = record_mandate(
|
|
session,
|
|
principal,
|
|
definition=definition_from_mapping(payload.definition),
|
|
expected_revision=payload.expected_revision,
|
|
)
|
|
session.commit()
|
|
except (MandateStoreError, InstitutionalContextError) as exc:
|
|
session.rollback()
|
|
raise _error(exc) from exc
|
|
return item.to_dict(include_inspection=True)
|
|
|
|
|
|
@router.post("/resolve", response_model=dict[str, Any])
|
|
def api_resolve_mandate(
|
|
payload: MandateResolutionPayload,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> dict[str, Any]:
|
|
_require_scope(principal, READ_SCOPE)
|
|
try:
|
|
result = SqlMandateResolver().resolve_mandate(
|
|
session,
|
|
principal,
|
|
request=resolution_request_from_mapping(payload.request),
|
|
)
|
|
except (MandateStoreError, InstitutionalContextError) as exc:
|
|
raise _error(exc) from exc
|
|
return result.to_dict(include_inspection=True)
|
|
|
|
|
|
__all__ = ["router"]
|