Extract HTML5 response helpers

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

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi.responses import Response, StreamingResponse
+from fastapi.staticfiles import StaticFiles
+
+HTML5_SECURITY_HEADERS = {"X-Content-Type-Options": "nosniff"}
+
+
+class Html5StaticFiles(StaticFiles):
+    def file_response(self, *args, **kwargs):
+        response = super().file_response(*args, **kwargs)
+        response.headers.setdefault("Cache-Control", "public, max-age=86400")
+        response.headers.setdefault("X-Content-Type-Options", "nosniff")
+        return response
+
+
+def html5_sse_headers() -> dict[str, str]:
+    return {
+        "Cache-Control": "no-cache, no-transform",
+        "Connection": "keep-alive",
+        "X-Accel-Buffering": "no",
+        **HTML5_SECURITY_HEADERS,
+    }
+
+
+def html5_response(fragment: str) -> Response:
+    return Response(
+        fragment,
+        media_type="text/html; charset=utf-8",
+        headers={"Cache-Control": "no-cache, no-transform", **HTML5_SECURITY_HEADERS},
+    )
+
+
+def html5_sse_response(content: Any) -> StreamingResponse:
+    return StreamingResponse(
+        content,
+        media_type="text/event-stream",
+        headers=html5_sse_headers(),
+    )

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

@@ -33,11 +33,15 @@ from collaboration import (
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import PlainTextResponse, Response, StreamingResponse
-from fastapi.staticfiles import StaticFiles
 from neo4j import AsyncGraphDatabase
 from pydantic import BaseModel, Field
 
 from api_server.html5_projects import render_html5_index, render_html5_project_rows
+from api_server.html5_responses import (
+    Html5StaticFiles,
+    html5_response as _html5_response,
+    html5_sse_response as _html5_sse_response,
+)
 from api_server.html5_inspector import (
     render_html5_flowchart,
     render_html5_object_context,
@@ -135,15 +139,6 @@ from ui_semantics import form_semantics
 
 app = FastAPI(title="SFERA API", version="0.1.0")
 _HTML5_ASSETS_DIR = Path(__file__).resolve().parent / "static" / "html5"
-_HTML5_SECURITY_HEADERS = {"X-Content-Type-Options": "nosniff"}
-
-
-class Html5StaticFiles(StaticFiles):
-    def file_response(self, *args, **kwargs):
-        response = super().file_response(*args, **kwargs)
-        response.headers.setdefault("Cache-Control", "public, max-age=86400")
-        response.headers.setdefault("X-Content-Type-Options", "nosniff")
-        return response
 
 
 app.mount("/html5/assets", Html5StaticFiles(directory=_HTML5_ASSETS_DIR), name="html5-assets")
@@ -8435,31 +8430,6 @@ def _html5_sse_event(event: str, fragment: str) -> str:
     return f"event: {event}\nretry: 5000\n{data}\n\n"
 
 
-def _html5_sse_headers() -> dict[str, str]:
-    return {
-        "Cache-Control": "no-cache, no-transform",
-        "Connection": "keep-alive",
-        "X-Accel-Buffering": "no",
-        **_HTML5_SECURITY_HEADERS,
-    }
-
-
-def _html5_response(fragment: str) -> Response:
-    return Response(
-        fragment,
-        media_type="text/html; charset=utf-8",
-        headers={"Cache-Control": "no-cache, no-transform", **_HTML5_SECURITY_HEADERS},
-    )
-
-
-def _html5_sse_response(content: Any) -> StreamingResponse:
-    return StreamingResponse(
-        content,
-        media_type="text/event-stream",
-        headers=_html5_sse_headers(),
-    )
-
-
 def _html5_sse_if_changed(last_fragments: dict[str, str], event: str, fragment: str):
     if last_fragments.get(event) == fragment:
         return