Extract HTML5 setup controller

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

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator, Callable, Iterable
+from typing import Any
+
+from fastapi import HTTPException
+
+from api_server.html5_forms import form_value
+from api_server.html5_setup import (
+    render_html5_import_check,
+    render_html5_import_job,
+    render_html5_project_setup,
+    render_html5_settings_panel,
+    render_html5_setup_summary,
+)
+from api_server.html5_sse import html5_sse_comment, html5_sse_if_changed
+
+
+def html5_setup_page(
+    *,
+    project_id: str,
+    projects: Iterable[object],
+    setup: object,
+) -> str:
+    return render_html5_project_setup(project_id=project_id, projects=projects, setup=setup)
+
+
+def html5_setup_summary(*, project_id: str, setup: object) -> str:
+    return render_html5_setup_summary(project_id, setup)
+
+
+async def html5_setup_event_stream(
+    *,
+    project_id: str,
+    setup_response: Callable[[str], object],
+    latest_import_job: Callable[[str], object | None],
+    once: bool = False,
+) -> AsyncIterator[str]:
+    last_fragments: dict[str, str] = {}
+    while True:
+        yield html5_sse_comment(f"setup {project_id} heartbeat")
+        try:
+            setup = setup_response(project_id)
+        except HTTPException as error:
+            setup_error = f'<div class="setup-summary" data-html5-setup-summary><p class="muted padded">{error.detail}</p></div>'
+            for event_text in html5_sse_if_changed(last_fragments, "setup-summary", setup_error):
+                yield event_text
+            if once:
+                break
+            await asyncio.sleep(2)
+            continue
+        for event_text in html5_sse_if_changed(last_fragments, "setup-summary", render_html5_setup_summary(project_id, setup)):
+            yield event_text
+        for event_text in html5_sse_if_changed(
+            last_fragments,
+            "setup-import-job",
+            render_html5_import_job(project_id, latest_import_job(project_id)),
+        ):
+            yield event_text
+        if once:
+            break
+        await asyncio.sleep(2)
+
+
+async def html5_setup_source(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    import_source_kind: Callable[[str], Any],
+    setup_response: Callable[[str], object],
+    save_settings: Callable[[str, object], Any],
+) -> str:
+    source = import_source_kind(form_value(form, "source") or "XML_DUMP")
+    current = setup_response(project_id)
+    settings = current.settings.model_copy(update={"structure_source": source})
+    setup = await save_settings(project_id, settings)
+    return render_html5_setup_summary(project_id, setup)
+
+
+async def html5_setup_settings(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    setup_response: Callable[[str], object],
+    save_settings: Callable[[str, object], Any],
+) -> str:
+    current = setup_response(project_id)
+    settings = current.settings.model_copy(
+        update={
+            "name": form_value(form, "name") or current.settings.name,
+            "platform_version": form_value(form, "platform_version"),
+            "compatibility_mode": form_value(form, "compatibility_mode"),
+        }
+    )
+    setup = await save_settings(project_id, settings)
+    return render_html5_settings_panel(project_id, setup, saved=True)
+
+
+def html5_setup_check(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    import_source_kind: Callable[[str], Any],
+    import_request: Callable[..., object],
+    current_import_source: Callable[[str], object],
+    import_check: Callable[[str, object, object], object],
+) -> str:
+    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
+    check = import_check(project_id, source, import_request(source=source))
+    return render_html5_import_check(project_id, check)
+
+
+def html5_setup_import(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    import_source_kind: Callable[[str], Any],
+    import_request: Callable[..., object],
+    current_import_source: Callable[[str], object],
+    execute_import: Callable[[str, object], object],
+    setup_response: Callable[[str], object],
+) -> str:
+    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
+    structure_only = form_value(form, "structure_only") in {"1", "true", "on", "yes"}
+    execute_import(project_id, import_request(source=source, structure_only=structure_only))
+    return render_html5_setup_summary(project_id, setup_response(project_id))
+
+
+async def html5_setup_import_job(
+    *,
+    project_id: str,
+    form: dict[str, list[str]],
+    import_source_kind: Callable[[str], Any],
+    import_request: Callable[..., object],
+    current_import_source: Callable[[str], object],
+    start_import_job: Callable[[str, object, object], Any],
+) -> str:
+    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
+    job = await start_import_job(project_id, source, import_request(source=source))
+    return render_html5_import_job(project_id, job)
+
+
+def html5_setup_job(
+    *,
+    project_id: str,
+    job_id: str,
+    jobs_by_id: dict[str, object],
+) -> str:
+    job = jobs_by_id.get(job_id)
+    if job is None or (getattr(job, "payload", {}) or {}).get("project_id") != project_id:
+        raise HTTPException(status_code=404, detail=f"Unknown import job: {job_id}")
+    return render_html5_import_job(project_id, job)
+
+
+async def html5_setup_reindex(
+    *,
+    project_id: str,
+    reindex: Callable[[str], Any],
+    setup_response: Callable[[str], object],
+) -> str:
+    await reindex(project_id)
+    return render_html5_setup_summary(project_id, setup_response(project_id))