497 lines
14 KiB
Python
497 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.audit.logging import audit_from_principal
|
|
from govoplan_core.auth import ApiPrincipal, require_any_scope
|
|
from govoplan_core.core.semantic_documentation import (
|
|
SemanticDocumentationContractError,
|
|
SemanticDocumentationSubjectQuery,
|
|
SemanticDocumentationSubjectReference,
|
|
list_semantic_documentation_subjects,
|
|
)
|
|
from govoplan_core.db.session import get_session
|
|
|
|
from govoplan_docs.backend.manifest import (
|
|
DOCS_ADMIN_READ_SCOPE,
|
|
DOCS_READ_SCOPE,
|
|
DOCS_SEMANTIC_CREATE_SCOPE,
|
|
DOCS_SEMANTIC_EDIT_SCOPE,
|
|
DOCS_SEMANTIC_EXPORT_SCOPE,
|
|
DOCS_SEMANTIC_POLICY_SCOPE,
|
|
DOCS_SEMANTIC_PUBLISH_SCOPE,
|
|
DOCS_SEMANTIC_RETIRE_SCOPE,
|
|
DOCS_SEMANTIC_SUPERSEDE_SCOPE,
|
|
)
|
|
from govoplan_docs.backend.semantic_schemas import (
|
|
SemanticDocumentationCreateRequest,
|
|
SemanticDocumentationPolicyUpdateRequest,
|
|
SemanticDocumentationSupersedeRequest,
|
|
SemanticDocumentationTransitionRequest,
|
|
SemanticDocumentationUpdateRequest,
|
|
)
|
|
from govoplan_docs.backend.semantic_service import (
|
|
SemanticDocumentationAuthorizationError,
|
|
SemanticDocumentationConflictError,
|
|
SemanticDocumentationError,
|
|
SemanticDocumentationNotFoundError,
|
|
content_visible_to_principal,
|
|
create_semantic_entry,
|
|
get_semantic_entry,
|
|
list_semantic_entries,
|
|
publication_policy,
|
|
publish_semantic_entry,
|
|
retire_semantic_entry,
|
|
revision_payload,
|
|
select_locale_entries,
|
|
semantic_entry_history,
|
|
semantic_entry_payload,
|
|
set_publication_policy,
|
|
supersede_semantic_entry,
|
|
update_semantic_entry,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/semantic", tags=["docs-semantic"])
|
|
READ_SCOPES = (
|
|
DOCS_READ_SCOPE,
|
|
DOCS_ADMIN_READ_SCOPE,
|
|
DOCS_SEMANTIC_CREATE_SCOPE,
|
|
DOCS_SEMANTIC_EDIT_SCOPE,
|
|
DOCS_SEMANTIC_PUBLISH_SCOPE,
|
|
DOCS_SEMANTIC_SUPERSEDE_SCOPE,
|
|
DOCS_SEMANTIC_RETIRE_SCOPE,
|
|
)
|
|
EDITOR_SCOPES = READ_SCOPES[2:]
|
|
|
|
SessionDep = Annotated[Session, Depends(get_session)]
|
|
ReadPrincipal = Annotated[ApiPrincipal, Depends(require_any_scope(*READ_SCOPES))]
|
|
|
|
|
|
@router.get("/policy")
|
|
def get_policy(
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: ReadPrincipal,
|
|
) -> dict[str, str]:
|
|
_private(response)
|
|
return {"mode": publication_policy(session, principal.tenant_id)}
|
|
|
|
|
|
@router.put("/policy")
|
|
def update_policy(
|
|
payload: SemanticDocumentationPolicyUpdateRequest,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_POLICY_SCOPE))
|
|
],
|
|
) -> dict[str, str]:
|
|
_private(response)
|
|
try:
|
|
mode = set_publication_policy(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
mode=payload.mode,
|
|
)
|
|
_audit(session, principal, "docs.semantic.policy.updated", None, {"mode": mode})
|
|
session.commit()
|
|
return {"mode": mode}
|
|
except SemanticDocumentationError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.get("/subjects")
|
|
def list_subjects(
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: ReadPrincipal,
|
|
query: str = Query(default="", max_length=300),
|
|
subject_kind: list[str] | None = Query(default=None, max_length=120),
|
|
module_id: str | None = Query(default=None, max_length=80),
|
|
limit: int = Query(default=50, ge=1, le=200),
|
|
cursor: str | None = Query(default=None, max_length=1000),
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
try:
|
|
pages = list_semantic_documentation_subjects(
|
|
_registry(request),
|
|
session,
|
|
principal,
|
|
request=SemanticDocumentationSubjectQuery(
|
|
tenant_id=principal.tenant_id,
|
|
query=query,
|
|
subject_kinds=tuple(subject_kind or ()),
|
|
limit=limit,
|
|
cursor=cursor,
|
|
),
|
|
)
|
|
except (SemanticDocumentationContractError, TypeError) as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
items = [
|
|
{
|
|
"module_id": provider_module_id,
|
|
"subjects": [item.to_dict() for item in page.subjects],
|
|
"next_cursor": page.next_cursor,
|
|
"has_more": page.has_more,
|
|
}
|
|
for provider_module_id, page in pages
|
|
if module_id is None or provider_module_id == module_id
|
|
]
|
|
return {"providers": items, "total": sum(len(item["subjects"]) for item in items)}
|
|
|
|
|
|
@router.get("/entries")
|
|
def list_entries(
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: ReadPrincipal,
|
|
locale: str = Query(default="de", min_length=2, max_length=20),
|
|
module_id: str | None = Query(default=None, max_length=80),
|
|
subject_kind: str | None = Query(default=None, max_length=120),
|
|
include_drafts: bool = Query(default=False),
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
editor = include_drafts and _has_any(principal, EDITOR_SCOPES)
|
|
entries = list_semantic_entries(
|
|
session,
|
|
principal,
|
|
module_id=module_id,
|
|
subject_kind=subject_kind,
|
|
)
|
|
selected = entries if editor else select_locale_entries(entries, locale=locale)
|
|
items = [
|
|
item
|
|
for entry in selected
|
|
if (
|
|
item := semantic_entry_payload(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
entry=entry,
|
|
editor=editor,
|
|
requested_locale=locale,
|
|
)
|
|
)
|
|
is not None
|
|
]
|
|
return {"items": items, "total": len(items)}
|
|
|
|
|
|
@router.get("/entries/{entry_id}")
|
|
def inspect_entry(
|
|
entry_id: str,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: ReadPrincipal,
|
|
include_draft: bool = Query(default=False),
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
try:
|
|
entry = get_semantic_entry(session, principal, entry_id=entry_id)
|
|
item = semantic_entry_payload(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
entry=entry,
|
|
editor=include_draft and _has_any(principal, EDITOR_SCOPES),
|
|
)
|
|
if item is None:
|
|
raise SemanticDocumentationNotFoundError("Entry not found.")
|
|
return item
|
|
except SemanticDocumentationError as exc:
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.get("/entries/{entry_id}/history")
|
|
def inspect_history(
|
|
entry_id: str,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[ApiPrincipal, Depends(require_any_scope(*EDITOR_SCOPES))],
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
try:
|
|
items = semantic_entry_history(session, principal, entry_id=entry_id)
|
|
return {"items": [revision_payload(item) for item in items], "total": len(items)}
|
|
except SemanticDocumentationError as exc:
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.post("/entries", status_code=status.HTTP_201_CREATED)
|
|
def create_entry(
|
|
payload: SemanticDocumentationCreateRequest,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_CREATE_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
return _mutate(
|
|
response,
|
|
session,
|
|
principal,
|
|
action="created",
|
|
callback=lambda: create_semantic_entry(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
subject=SemanticDocumentationSubjectReference.from_mapping(
|
|
payload.subject.model_dump(mode="json")
|
|
),
|
|
locale=payload.locale,
|
|
content=payload.content.model_dump(mode="json"),
|
|
change_reason=payload.change_reason,
|
|
),
|
|
registry=_registry(request),
|
|
)
|
|
|
|
|
|
@router.put("/entries/{entry_id}")
|
|
def update_entry(
|
|
entry_id: str,
|
|
payload: SemanticDocumentationUpdateRequest,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_EDIT_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
return _mutate(
|
|
response,
|
|
session,
|
|
principal,
|
|
action="updated",
|
|
callback=lambda: update_semantic_entry(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
entry_id=entry_id,
|
|
expected_revision=payload.expected_revision,
|
|
content=payload.content.model_dump(mode="json"),
|
|
change_reason=payload.change_reason,
|
|
),
|
|
registry=_registry(request),
|
|
)
|
|
|
|
|
|
@router.post("/entries/{entry_id}/publish")
|
|
def publish_entry(
|
|
entry_id: str,
|
|
payload: SemanticDocumentationTransitionRequest,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_PUBLISH_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
return _mutate(
|
|
response,
|
|
session,
|
|
principal,
|
|
action="published",
|
|
callback=lambda: publish_semantic_entry(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
entry_id=entry_id,
|
|
expected_revision=payload.expected_revision,
|
|
change_reason=payload.change_reason,
|
|
),
|
|
registry=_registry(request),
|
|
)
|
|
|
|
|
|
@router.post("/entries/{entry_id}/supersede")
|
|
def supersede_entry(
|
|
entry_id: str,
|
|
payload: SemanticDocumentationSupersedeRequest,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_SUPERSEDE_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
return _mutate(
|
|
response,
|
|
session,
|
|
principal,
|
|
action="superseded",
|
|
callback=lambda: supersede_semantic_entry(
|
|
session,
|
|
principal,
|
|
entry_id=entry_id,
|
|
replacement_entry_id=payload.replacement_entry_id,
|
|
expected_revision=payload.expected_revision,
|
|
change_reason=payload.change_reason,
|
|
),
|
|
registry=_registry(request),
|
|
)
|
|
|
|
|
|
@router.post("/entries/{entry_id}/retire")
|
|
def retire_entry(
|
|
entry_id: str,
|
|
payload: SemanticDocumentationTransitionRequest,
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_RETIRE_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
return _mutate(
|
|
response,
|
|
session,
|
|
principal,
|
|
action="retired",
|
|
callback=lambda: retire_semantic_entry(
|
|
session,
|
|
principal,
|
|
entry_id=entry_id,
|
|
expected_revision=payload.expected_revision,
|
|
change_reason=payload.change_reason,
|
|
),
|
|
registry=_registry(request),
|
|
)
|
|
|
|
|
|
@router.get("/export")
|
|
def export_tenant_semantics(
|
|
request: Request,
|
|
response: Response,
|
|
session: SessionDep,
|
|
principal: Annotated[
|
|
ApiPrincipal, Depends(require_any_scope(DOCS_SEMANTIC_EXPORT_SCOPE))
|
|
],
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
entries = list_semantic_entries(session, principal)
|
|
response.headers["Content-Disposition"] = (
|
|
'attachment; filename="govoplan-semantic-documentation.json"'
|
|
)
|
|
exported: list[dict[str, object]] = []
|
|
for entry in entries:
|
|
payload = semantic_entry_payload(
|
|
session,
|
|
_registry(request),
|
|
principal,
|
|
entry=entry,
|
|
editor=True,
|
|
)
|
|
if payload is None or payload.get("content_redacted"):
|
|
continue
|
|
exported.append(
|
|
{
|
|
"entry": payload,
|
|
"history": [
|
|
revision_payload(item)
|
|
for item in semantic_entry_history(
|
|
session, principal, entry_id=entry.id
|
|
)
|
|
if content_visible_to_principal(item.content, principal)
|
|
],
|
|
}
|
|
)
|
|
return {
|
|
"schema_version": "1",
|
|
"tenant_id": principal.tenant_id,
|
|
"entries": exported,
|
|
}
|
|
|
|
|
|
def _mutate(
|
|
response: Response,
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
action: str,
|
|
callback: Callable[[], Any],
|
|
registry: object,
|
|
) -> dict[str, object]:
|
|
_private(response)
|
|
try:
|
|
entry = callback()
|
|
_audit(
|
|
session,
|
|
principal,
|
|
f"docs.semantic.{action}",
|
|
entry.id,
|
|
{
|
|
"revision": entry.current_revision,
|
|
"subject_stable_key": entry.subject_stable_key,
|
|
"locale": entry.locale,
|
|
},
|
|
)
|
|
session.commit()
|
|
result = semantic_entry_payload(
|
|
session,
|
|
registry,
|
|
principal,
|
|
entry=entry,
|
|
editor=True,
|
|
)
|
|
if result is None:
|
|
raise SemanticDocumentationNotFoundError("Entry not found.")
|
|
return result
|
|
except (SemanticDocumentationError, SemanticDocumentationContractError) as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
def _audit(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
action: str,
|
|
entry_id: str | None,
|
|
details: dict[str, object],
|
|
) -> None:
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action=action,
|
|
object_type="semantic_documentation",
|
|
object_id=entry_id,
|
|
details=details,
|
|
)
|
|
|
|
|
|
def _private(response: Response) -> None:
|
|
response.headers["Cache-Control"] = "private, no-store"
|
|
|
|
|
|
def _registry(request: Request) -> object:
|
|
registry = getattr(request.app.state, "govoplan_registry", None)
|
|
if registry is None or not hasattr(registry, "capability_names"):
|
|
raise HTTPException(status_code=500, detail="Module registry is unavailable.")
|
|
return registry
|
|
|
|
|
|
def _has_any(principal: ApiPrincipal, scopes: tuple[str, ...]) -> bool:
|
|
return any(principal.has(scope) for scope in scopes)
|
|
|
|
|
|
def _http_error(exc: Exception) -> HTTPException:
|
|
if isinstance(exc, SemanticDocumentationNotFoundError):
|
|
return HTTPException(status_code=404, detail=str(exc))
|
|
if isinstance(exc, SemanticDocumentationAuthorizationError):
|
|
return HTTPException(status_code=403, detail=str(exc))
|
|
if isinstance(exc, SemanticDocumentationConflictError):
|
|
return HTTPException(status_code=409, detail=str(exc))
|
|
return HTTPException(status_code=422, detail=str(exc))
|
|
|
|
|
|
__all__ = ["router"]
|