SharedDocsLib/docslib/server.py

274 lines
9 KiB
Python

from __future__ import annotations
import importlib
import json
import runpy
import sys
from pathlib import Path
from threading import RLock
from typing import Mapping, Sequence
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse, Response
from .compiler import BuildResult, Compiler
from .registry import registry
from .ui import HeaderButton
NO_CACHE_HEADERS = {
"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
"Pragma": "no-cache",
"Expires": "0",
}
def _manifest_url(manifest_path: str | None, manifest_parh: str | None) -> str:
if manifest_path and manifest_parh:
raise ValueError("Use either manifest_path or the legacy typo manifest_parh, not both")
value = manifest_path or manifest_parh or "./manifest.json"
if value.endswith(".json"):
return value
return value.rstrip("/") + "/manifest.json"
class DocumentationSite:
def __init__(self, compiler: Compiler, *, live_preview: bool) -> None:
self.compiler = compiler
self.live_preview = live_preview
self._lock = RLock()
self._result = compiler.build()
self._last_error: str | None = None
self._source_modules: dict[Path, set[str]] = {}
self._source_mtimes: dict[Path, int] = {}
self._discover_sources()
@property
def result(self) -> BuildResult:
with self._lock:
return self._result
@property
def last_error(self) -> str | None:
with self._lock:
return self._last_error
def _discover_sources(self) -> None:
for page in registry.pages():
if not page.source_file:
continue
path = Path(page.source_file).resolve()
self._source_modules.setdefault(path, set()).add(page.module)
try:
self._source_mtimes[path] = path.stat().st_mtime_ns
except OSError:
continue
for path in self.compiler.watched_files():
try:
self._source_mtimes.setdefault(path, path.stat().st_mtime_ns)
except OSError:
continue
def refresh_changed_sources(self) -> bool:
if not self.live_preview:
return False
with self._lock:
changed = []
for path, previous in self._source_mtimes.items():
try:
current = path.stat().st_mtime_ns
except OSError:
continue
if current != previous:
changed.append((path, current))
if not changed:
return False
snapshot = registry.snapshot()
try:
for path, _ in changed:
self._reload_source(path)
new_result = self.compiler.build()
except Exception as error:
registry.restore(snapshot)
self._last_error = f"{type(error).__name__}: {error}"
for path, mtime in changed:
self._source_mtimes[path] = mtime
return False
self._result = new_result
self._last_error = None
for path, mtime in changed:
self._source_mtimes[path] = mtime
self._source_modules.clear()
self._discover_sources()
return True
def _reload_source(self, path: Path) -> None:
modules = self._source_modules.get(path, set())
if not modules:
return
registry.remove_source(path)
reloadable = [name for name in modules if name not in {"__main__", "<run_path>"} and name in sys.modules]
if reloadable:
for name in reloadable:
importlib.reload(sys.modules[name])
else:
# run_name is deliberately not __main__, so a guarded run() call is not executed again.
runpy.run_path(str(path), run_name=f"_docslib_live_{abs(hash(path))}")
def create_app(
*,
title: str = "Documentation",
assets_dir: str | Path = "./assets",
custom_css: str | Path | None = None,
custom_js: str | Path | None = None,
template_dir: str | Path | None = None,
manifest_path: str | None = None,
manifest_parh: str | None = None,
live_preview: bool = False,
header_buttons: Sequence[HeaderButton | Mapping[str, object]] | None = None,
cors_origins: Sequence[str] = ("*",),
) -> FastAPI:
"""Compile registered pages and return a ready-to-serve FastAPI application."""
manifest_url = _manifest_url(manifest_path, manifest_parh)
compiler = Compiler(
title=title,
assets_dir=assets_dir,
custom_css=custom_css,
custom_js=custom_js,
template_dir=template_dir,
manifest_url=manifest_url,
live_preview=live_preview,
header_buttons=header_buttons,
)
site = DocumentationSite(compiler, live_preview=live_preview)
app = FastAPI(title=f"{title} · SharedDocsLib", docs_url=None, redoc_url=None)
app.state.docslib = site
app.add_middleware(
CORSMiddleware,
allow_origins=list(cors_origins),
allow_credentials=False,
allow_methods=["GET"],
allow_headers=["*"],
)
@app.get("/", include_in_schema=False)
@app.get("/index.html", response_class=HTMLResponse)
def index() -> HTMLResponse:
return HTMLResponse(site.result.html)
@app.get("/manifest.json", response_class=JSONResponse)
def manifest() -> JSONResponse:
return JSONResponse(site.result.manifest, headers=NO_CACHE_HEADERS)
@app.get("/version")
def version() -> Response:
return Response(
str(site.result.manifest["revision"]),
media_type="text/plain",
headers=NO_CACHE_HEADERS,
)
@app.get("/index.zip")
def index_zip() -> Response:
return Response(
site.result.zip_archive,
media_type="application/zip",
headers={
"Content-Disposition": 'attachment; filename="index.zip"',
"Cache-Control": "no-store",
},
)
if live_preview:
@app.get("/live")
def live() -> JSONResponse:
changed = site.refresh_changed_sources()
result = site.result
return JSONResponse(
{
"enabled": True,
"changed": changed,
"revision": result.manifest["revision"],
"updated_at": result.manifest["updated_at"],
"error": site.last_error,
},
headers={"Cache-Control": "no-store"},
)
return app
def build(
*,
build_directory: str | Path = "./build",
title: str = "Documentation",
assets_dir: str | Path = "./assets",
custom_css: str | Path | None = None,
custom_js: str | Path | None = None,
template_dir: str | Path | None = None,
manifest_path: str | None = None,
manifest_parh: str | None = None,
header_buttons: Sequence[HeaderButton | Mapping[str, object]] | None = None,
) -> BuildResult:
"""Compile registered pages and write a static documentation bundle."""
compiler = Compiler(
title=title,
assets_dir=assets_dir,
custom_css=custom_css,
custom_js=custom_js,
template_dir=template_dir,
manifest_url=_manifest_url(manifest_path, manifest_parh),
live_preview=False,
header_buttons=header_buttons,
)
result = compiler.build()
output = Path(build_directory).expanduser().resolve()
output.mkdir(parents=True, exist_ok=True)
(output / "index.html").write_bytes(result.html)
(output / "manifest.json").write_text(
json.dumps(result.manifest, ensure_ascii=False, separators=(",", ":")),
encoding="utf-8",
)
(output / "version").write_text(str(result.manifest["revision"]), encoding="utf-8")
(output / "index.zip").write_bytes(result.zip_archive)
return result
def run(
*,
port: int = 8000,
host: str = "127.0.0.1",
title: str = "Documentation",
assets_dir: str | Path = "./assets",
custom_css: str | Path | None = None,
custom_js: str | Path | None = None,
template_dir: str | Path | None = None,
manifest_path: str | None = None,
manifest_parh: str | None = None,
live_preview: bool = False,
header_buttons: Sequence[HeaderButton | Mapping[str, object]] | None = None,
cors_origins: Sequence[str] = ("*",),
log_level: str = "info",
) -> None:
"""Compile the documentation and run it with Uvicorn."""
app = create_app(
title=title,
assets_dir=assets_dir,
custom_css=custom_css,
custom_js=custom_js,
template_dir=template_dir,
manifest_path=manifest_path,
manifest_parh=manifest_parh,
live_preview=live_preview,
header_buttons=header_buttons,
cors_origins=cors_origins,
)
uvicorn.run(app, host=host, port=port, log_level=log_level)
__all__ = ["DocumentationSite", "build", "create_app", "run"]