153 lines
4.4 KiB
Python
153 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from auth import require_admin_dep
|
|
from journal_generation_policy import (
|
|
PURPOSE_JOURNAL,
|
|
CatalogError,
|
|
GenerationPolicyError,
|
|
archive_guideline,
|
|
clone_guideline,
|
|
create_guideline,
|
|
default_selection_ids,
|
|
delete_guideline,
|
|
get_guideline,
|
|
overview_payload,
|
|
preview_selection,
|
|
publish_guideline,
|
|
reset_seed_drafts,
|
|
set_default_guideline,
|
|
update_guideline,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/admin/generation-instructions", tags=["admin"])
|
|
|
|
|
|
class GuidelineWrite(BaseModel):
|
|
slot: str | None = None
|
|
guideline_key: str | None = None
|
|
label: str | None = None
|
|
summary: str | None = None
|
|
instruction: str | None = None
|
|
style_context: dict | None = None
|
|
sort_order: int | None = None
|
|
|
|
|
|
class PreviewRequest(BaseModel):
|
|
generation_selection: dict | None = None
|
|
|
|
|
|
def _http(exc: GenerationPolicyError):
|
|
raise HTTPException(
|
|
status_code=exc.status_code,
|
|
detail={"code": exc.code, "message": exc.message},
|
|
) from exc
|
|
|
|
|
|
def _purpose(purpose: str) -> str:
|
|
if purpose != PURPOSE_JOURNAL:
|
|
raise HTTPException(404, "Unbekannter Generation-Purpose")
|
|
return purpose
|
|
|
|
|
|
@router.get("/{purpose}")
|
|
def list_overview(purpose: str, session: dict = Depends(require_admin_dep)):
|
|
return overview_payload(_purpose(purpose))
|
|
|
|
|
|
@router.post("/{purpose}/reset")
|
|
def reset_items(purpose: str, session: dict = Depends(require_admin_dep)):
|
|
try:
|
|
return reset_seed_drafts(_purpose(purpose))
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}/preview")
|
|
def preview_items(purpose: str, body: PreviewRequest, session: dict = Depends(require_admin_dep)):
|
|
checked = _purpose(purpose)
|
|
selection = body.generation_selection or default_selection_ids(checked)
|
|
try:
|
|
return preview_selection(selection, purpose=checked)
|
|
except GenerationPolicyError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}")
|
|
def create_item(purpose: str, body: GuidelineWrite, session: dict = Depends(require_admin_dep)):
|
|
try:
|
|
return create_guideline(_purpose(purpose), body.slot or "", body.model_dump())
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.get("/{purpose}/{guideline_id}")
|
|
def read_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
item = get_guideline(guideline_id, include_instruction=True)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
if item.get("purpose") != purpose:
|
|
raise HTTPException(404, "Ausprägung nicht gefunden.")
|
|
return item
|
|
|
|
|
|
@router.put("/{purpose}/{guideline_id}")
|
|
def write_item(purpose: str, guideline_id: str, body: GuidelineWrite, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
payload = {key: value for key, value in body.model_dump().items() if value is not None}
|
|
payload.pop("slot", None)
|
|
return update_guideline(guideline_id, payload)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}/{guideline_id}/clone")
|
|
def clone_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
return clone_guideline(guideline_id)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}/{guideline_id}/publish")
|
|
def publish_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
return publish_guideline(guideline_id)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}/{guideline_id}/archive")
|
|
def archive_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
return archive_guideline(guideline_id)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.post("/{purpose}/{guideline_id}/default")
|
|
def default_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
return set_default_guideline(guideline_id)
|
|
except CatalogError as exc:
|
|
_http(exc)
|
|
|
|
|
|
@router.delete("/{purpose}/{guideline_id}")
|
|
def delete_item(purpose: str, guideline_id: str, session: dict = Depends(require_admin_dep)):
|
|
_purpose(purpose)
|
|
try:
|
|
delete_guideline(guideline_id)
|
|
return {"ok": True}
|
|
except CatalogError as exc:
|
|
_http(exc)
|