Extract HTML5 authoring controller

services/api-server/src/api_server/html5_authoring_controller.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from typing import Any
+
+from fastapi import HTTPException
+
+from api_server.html5_authoring import (
+    render_html5_authoring_apply_result,
+    render_html5_authoring_change_detail,
+    render_html5_authoring_changes,
+    render_html5_authoring_diff_result,
+    render_html5_authoring_preview_result,
+    render_html5_authoring_rollback_result,
+    render_html5_metadata_apply_result,
+    render_html5_metadata_preview_result,
+)
+from api_server.html5_forms import form_value, html5_metadata_payload, html5_metadata_request_payload
+
+
+def html5_authoring_changes(project_id: str, changes: Iterable[object]) -> str:
+    return render_html5_authoring_changes(project_id, changes)
+
+
+def html5_authoring_change_detail(project_id: str, rollback_preview: object) -> str:
+    return render_html5_authoring_change_detail(project_id, rollback_preview)
+
+
+async def html5_authoring_apply_rollback(
+    *,
+    project_id: str,
+    change_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    apply_rollback: Callable[[str, str, object], Any],
+) -> str:
+    payload = request_model(
+        expected_rollback_version_id=form_value(form, "expected_rollback_version_id") or "",
+        approved_by=form_value(form, "approved_by") or "",
+        approval_note=form_value(form, "approval_note"),
+        task_id=form_value(form, "task_id"),
+        session_id=form_value(form, "session_id"),
+    )
+    try:
+        result = await apply_rollback(project_id, change_id, payload)
+        return render_html5_authoring_rollback_result(project_id, result)
+    except HTTPException as error:
+        return render_html5_authoring_rollback_result(project_id, error=str(error.detail))
+
+
+async def html5_authoring_completion_preview(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    completion_preview: Callable[[str, object], Any],
+) -> str:
+    cursor_line = _optional_int(form_value(form, "cursor_line"))
+    payload = request_model(
+        object_name=form_value(form, "object_name"),
+        routine_name=form_value(form, "routine_name"),
+        cursor_line=cursor_line,
+        source_text=form_value(form, "source_text"),
+        intent=form_value(form, "intent") or "guarded-return",
+        user_id=form_value(form, "user_id"),
+    )
+    try:
+        preview = await completion_preview(project_id, payload)
+        return render_html5_authoring_preview_result(project_id, preview)
+    except HTTPException as error:
+        return render_html5_authoring_preview_result(project_id, error=str(error.detail))
+
+
+def html5_authoring_semantic_diff_preview(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    semantic_diff_preview: Callable[[str, object], object],
+) -> str:
+    payload = request_model(**_semantic_diff_payload(form))
+    try:
+        preview = semantic_diff_preview(project_id, payload)
+        return render_html5_authoring_diff_result(
+            project_id,
+            preview,
+            request_payload=_semantic_diff_payload_from_request(payload),
+        )
+    except HTTPException as error:
+        return render_html5_authoring_diff_result(project_id, error=str(error.detail))
+
+
+async def html5_authoring_apply_change_set(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    apply_change_set: Callable[[str, object], Any],
+) -> str:
+    payload = request_model(
+        **_semantic_diff_payload(form),
+        expected_next_version_id=form_value(form, "expected_next_version_id") or "",
+        approved_by=form_value(form, "approved_by") or "",
+        approval_note=form_value(form, "approval_note"),
+    )
+    try:
+        result = await apply_change_set(project_id, payload)
+        return render_html5_authoring_apply_result(project_id, result)
+    except HTTPException as error:
+        return render_html5_authoring_apply_result(project_id, error=str(error.detail))
+
+
+def html5_authoring_metadata_object_preview(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    metadata_preview: Callable[[str, object], object],
+) -> str:
+    raw_payload = html5_metadata_payload(form)
+    payload = request_model(**html5_metadata_request_payload(raw_payload))
+    try:
+        preview = metadata_preview(project_id, payload)
+        return render_html5_metadata_preview_result(project_id, preview, request_payload=raw_payload)
+    except (HTTPException, ValueError) as error:
+        detail = getattr(error, "detail", str(error))
+        return render_html5_metadata_preview_result(project_id, error=str(detail))
+
+
+async def html5_authoring_apply_metadata_object(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    request_model: Callable[..., object],
+    apply_metadata_object: Callable[[str, object], Any],
+) -> str:
+    raw_payload = html5_metadata_payload(form)
+    payload = request_model(
+        **html5_metadata_request_payload(raw_payload),
+        expected_next_version_id=form_value(form, "expected_next_version_id") or "",
+        approved_by=form_value(form, "approved_by") or "",
+        approval_note=form_value(form, "approval_note"),
+    )
+    try:
+        result = await apply_metadata_object(project_id, payload)
+        return render_html5_metadata_apply_result(project_id, result)
+    except (HTTPException, ValueError) as error:
+        detail = getattr(error, "detail", str(error))
+        return render_html5_metadata_apply_result(project_id, error=str(detail))
+
+
+def _semantic_diff_payload(form: dict[str, list[str]]) -> dict[str, object]:
+    return {
+        "routine_name": form_value(form, "routine_name"),
+        "source_path": form_value(form, "source_path"),
+        "original_text": form_value(form, "original_text") or "",
+        "proposed_text": form_value(form, "proposed_text") or "",
+        "task_id": form_value(form, "task_id"),
+        "session_id": form_value(form, "session_id"),
+        "user_id": form_value(form, "user_id"),
+    }
+
+
+def _semantic_diff_payload_from_request(payload: object) -> dict[str, object]:
+    return {
+        "routine_name": getattr(payload, "routine_name", None),
+        "source_path": getattr(payload, "source_path", None),
+        "original_text": getattr(payload, "original_text", ""),
+        "proposed_text": getattr(payload, "proposed_text", ""),
+        "task_id": getattr(payload, "task_id", None),
+        "session_id": getattr(payload, "session_id", None),
+        "user_id": getattr(payload, "user_id", None),
+    }
+
+
+def _optional_int(value: str | None) -> int | None:
+    try:
+        return int(value) if value else None
+    except ValueError:
+        return None

services/api-server/src/api_server/main.py

@@ -39,8 +39,6 @@ from pydantic import BaseModel, Field
 from api_server.html5_forms import (
     form_value as _form_value,
     html5_form_data as _html5_form_data,
-    html5_metadata_payload as _html5_metadata_payload,
-    html5_metadata_request_payload as _html5_metadata_request_payload,
 )
 from api_server.html5_editor_controller import (
     html5_editor_page as _html5_editor_page,
@@ -59,20 +57,19 @@ from api_server.html5_sse import (
 )
 from api_server.html5_inspector import (
     render_html5_flowchart,
-    render_html5_object_context,
-    render_html5_object_report,
     render_html5_project_report,
     render_html5_review,
 )
-from api_server.html5_authoring import (
-    render_html5_authoring_apply_result,
-    render_html5_authoring_changes,
-    render_html5_authoring_change_detail,
-    render_html5_authoring_diff_result,
-    render_html5_authoring_preview_result,
-    render_html5_authoring_rollback_result,
-    render_html5_metadata_apply_result,
-    render_html5_metadata_preview_result,
+from api_server.html5_authoring import render_html5_authoring_changes
+from api_server.html5_authoring_controller import (
+    html5_authoring_apply_change_set as _html5_authoring_apply_change_set,
+    html5_authoring_apply_metadata_object as _html5_authoring_apply_metadata_object,
+    html5_authoring_apply_rollback as _html5_authoring_apply_rollback,
+    html5_authoring_change_detail as _html5_authoring_change_detail,
+    html5_authoring_changes as _html5_authoring_changes,
+    html5_authoring_completion_preview as _html5_authoring_completion_preview,
+    html5_authoring_metadata_object_preview as _html5_authoring_metadata_object_preview,
+    html5_authoring_semantic_diff_preview as _html5_authoring_semantic_diff_preview,
 )
 from api_server.html5_editor import (
     render_html5_source,
@@ -1880,146 +1877,91 @@ async def html5_project_object_context(project_id: str, object_name: str, mode:
 
 @app.get("/html5/projects/{project_id}/authoring/changes")
 async def html5_project_authoring_changes(project_id: str) -> Response:
-    return _html5_response(
-        render_html5_authoring_changes(project_id, _authoring_change_summaries(project_id)),
-    )
+    return _html5_response(_html5_authoring_changes(project_id, _authoring_change_summaries(project_id)))
 
 
 @app.get("/html5/projects/{project_id}/authoring/changes/{change_id}")
 async def html5_project_authoring_change_detail(project_id: str, change_id: str) -> Response:
-    return _html5_response(
-        render_html5_authoring_change_detail(project_id, _authoring_rollback_preview(project_id, change_id)),
-    )
+    return _html5_response(_html5_authoring_change_detail(project_id, _authoring_rollback_preview(project_id, change_id)))
 
 
 @app.post("/html5/projects/{project_id}/authoring/changes/{change_id}/apply-rollback")
 async def html5_project_authoring_apply_rollback(project_id: str, change_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    payload = AuthoringApplyRollbackRequest(
-        expected_rollback_version_id=_form_value(form, "expected_rollback_version_id") or "",
-        approved_by=_form_value(form, "approved_by") or "",
-        approval_note=_form_value(form, "approval_note"),
-        task_id=_form_value(form, "task_id"),
-        session_id=_form_value(form, "session_id"),
+    return _html5_response(
+        await _html5_authoring_apply_rollback(
+            project_id=project_id,
+            change_id=change_id,
+            form=form,
+            request_model=AuthoringApplyRollbackRequest,
+            apply_rollback=authoring_apply_rollback,
+        )
     )
-    try:
-        result = await authoring_apply_rollback(project_id, change_id, payload)
-        html = render_html5_authoring_rollback_result(project_id, result)
-    except HTTPException as error:
-        html = render_html5_authoring_rollback_result(project_id, error=str(error.detail))
-    return _html5_response(html)
 
 
 @app.post("/html5/projects/{project_id}/authoring/completion-preview")
 async def html5_project_authoring_completion_preview(project_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    cursor_line_value = _form_value(form, "cursor_line")
-    try:
-        cursor_line = int(cursor_line_value) if cursor_line_value else None
-    except ValueError:
-        cursor_line = None
-    payload = AuthoringCompletionPreviewRequest(
-        object_name=_form_value(form, "object_name"),
-        routine_name=_form_value(form, "routine_name"),
-        cursor_line=cursor_line,
-        source_text=_form_value(form, "source_text"),
-        intent=_form_value(form, "intent") or "guarded-return",
-        user_id=_form_value(form, "user_id"),
+    return _html5_response(
+        await _html5_authoring_completion_preview(
+            project_id=project_id,
+            form=form,
+            request_model=AuthoringCompletionPreviewRequest,
+            completion_preview=authoring_completion_preview,
+        )
     )
-    try:
-        preview = await authoring_completion_preview(project_id, payload)
-        html = render_html5_authoring_preview_result(project_id, preview)
-    except HTTPException as error:
-        html = render_html5_authoring_preview_result(project_id, error=str(error.detail))
-    return _html5_response(html)
 
 
 @app.post("/html5/projects/{project_id}/authoring/semantic-diff-preview")
 async def html5_project_authoring_semantic_diff_preview(project_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    payload = AuthoringSemanticDiffPreviewRequest(
-        routine_name=_form_value(form, "routine_name"),
-        source_path=_form_value(form, "source_path"),
-        original_text=_form_value(form, "original_text") or "",
-        proposed_text=_form_value(form, "proposed_text") or "",
-        task_id=_form_value(form, "task_id"),
-        session_id=_form_value(form, "session_id"),
-        user_id=_form_value(form, "user_id"),
+    return _html5_response(
+        _html5_authoring_semantic_diff_preview(
+            project_id=project_id,
+            form=form,
+            request_model=AuthoringSemanticDiffPreviewRequest,
+            semantic_diff_preview=_authoring_semantic_diff_preview,
         )
-    try:
-        preview = _authoring_semantic_diff_preview(project_id, payload)
-        html = render_html5_authoring_diff_result(
-            project_id,
-            preview,
-            request_payload={
-                "routine_name": payload.routine_name,
-                "source_path": payload.source_path,
-                "original_text": payload.original_text,
-                "proposed_text": payload.proposed_text,
-                "task_id": payload.task_id,
-                "session_id": payload.session_id,
-                "user_id": payload.user_id,
-            },
     )
-    except HTTPException as error:
-        html = render_html5_authoring_diff_result(project_id, error=str(error.detail))
-    return _html5_response(html)
 
 
 @app.post("/html5/projects/{project_id}/authoring/apply-change-set")
 async def html5_project_authoring_apply_change_set(project_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    payload = AuthoringApplyChangeSetRequest(
-        routine_name=_form_value(form, "routine_name"),
-        source_path=_form_value(form, "source_path"),
-        original_text=_form_value(form, "original_text") or "",
-        proposed_text=_form_value(form, "proposed_text") or "",
-        task_id=_form_value(form, "task_id"),
-        session_id=_form_value(form, "session_id"),
-        user_id=_form_value(form, "user_id"),
-        expected_next_version_id=_form_value(form, "expected_next_version_id") or "",
-        approved_by=_form_value(form, "approved_by") or "",
-        approval_note=_form_value(form, "approval_note"),
+    return _html5_response(
+        await _html5_authoring_apply_change_set(
+            project_id=project_id,
+            form=form,
+            request_model=AuthoringApplyChangeSetRequest,
+            apply_change_set=authoring_apply_change_set,
+        )
     )
-    try:
-        result = await authoring_apply_change_set(project_id, payload)
-        html = render_html5_authoring_apply_result(project_id, result)
-    except HTTPException as error:
-        html = render_html5_authoring_apply_result(project_id, error=str(error.detail))
-    return _html5_response(html)
 
 
 @app.post("/html5/projects/{project_id}/authoring/metadata-object-preview")
 async def html5_project_authoring_metadata_object_preview(project_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    raw_payload = _html5_metadata_payload(form)
-    payload = AuthoringMetadataObjectPreviewRequest(**_html5_metadata_request_payload(raw_payload))
-    try:
-        preview = _authoring_metadata_object_preview(project_id, payload)
-        html = render_html5_metadata_preview_result(project_id, preview, request_payload=raw_payload)
-    except (HTTPException, ValueError) as error:
-        detail = getattr(error, "detail", str(error))
-        html = render_html5_metadata_preview_result(project_id, error=str(detail))
-    return _html5_response(html)
+    return _html5_response(
+        _html5_authoring_metadata_object_preview(
+            project_id=project_id,
+            form=form,
+            request_model=AuthoringMetadataObjectPreviewRequest,
+            metadata_preview=_authoring_metadata_object_preview,
+        )
+    )
 
 
 @app.post("/html5/projects/{project_id}/authoring/apply-metadata-object")
 async def html5_project_authoring_apply_metadata_object(project_id: str, request: Request) -> Response:
     form = await _html5_form_data(request)
-    raw_payload = _html5_metadata_payload(form)
-    payload = AuthoringApplyMetadataObjectRequest(
-        **_html5_metadata_request_payload(raw_payload),
-        expected_next_version_id=_form_value(form, "expected_next_version_id") or "",
-        approved_by=_form_value(form, "approved_by") or "",
-        approval_note=_form_value(form, "approval_note"),
+    return _html5_response(
+        await _html5_authoring_apply_metadata_object(
+            project_id=project_id,
+            form=form,
+            request_model=AuthoringApplyMetadataObjectRequest,
+            apply_metadata_object=authoring_apply_metadata_object,
+        )
     )
-    try:
-        result = await authoring_apply_metadata_object(project_id, payload)
-        html = render_html5_metadata_apply_result(project_id, result)
-    except (HTTPException, ValueError) as error:
-        detail = getattr(error, "detail", str(error))
-        html = render_html5_metadata_apply_result(project_id, error=str(detail))
-    return _html5_response(html)
 
 
 @app.get("/html5/projects/{project_id}/setup")