278 lines
9.8 KiB
Python
278 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import runpy
|
|
import sys
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from types import ModuleType
|
|
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 .generators import DefaultGenerator, FilesBuildResult, FilesGenerator
|
|
from .hooks import VariantSelection
|
|
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:
|
|
self._source_modules.clear()
|
|
for source_file, module in registry.sources():
|
|
path = Path(source_file)
|
|
self._source_modules.setdefault(path, set()).add(module)
|
|
watched = self._source_modules.keys() | set(self.compiler.watched_files())
|
|
for path in self._source_mtimes.keys() - watched:
|
|
del self._source_mtimes[path]
|
|
for path in watched - self._source_mtimes.keys():
|
|
try:
|
|
self._source_mtimes[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()
|
|
module_snapshots: dict[ModuleType, dict[str, object]] = {}
|
|
try:
|
|
for path, _ in changed:
|
|
self._reload_source(path, module_snapshots)
|
|
new_result = self.compiler.build()
|
|
except Exception as error:
|
|
for module, namespace in module_snapshots.items():
|
|
module.__dict__.clear()
|
|
module.__dict__.update(namespace)
|
|
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._discover_sources()
|
|
return True
|
|
|
|
def _reload_source(self, path: Path, snapshots: dict[ModuleType, dict[str, object]]) -> 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:
|
|
# Read the source directly so a same-size edit within one second bypasses stale bytecode.
|
|
code = compile(path.read_bytes(), str(path), "exec", dont_inherit=True)
|
|
for name in reloadable:
|
|
module = sys.modules[name]
|
|
snapshots.setdefault(module, module.__dict__.copy())
|
|
exec(code, module.__dict__)
|
|
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,
|
|
component_variant: VariantSelection = 1,
|
|
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,
|
|
component_variant=component_variant,
|
|
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,
|
|
generator: DefaultGenerator | FilesGenerator = DefaultGenerator(),
|
|
) -> BuildResult | FilesBuildResult:
|
|
"""Generate a static HTML bundle or raw content files from registered pages."""
|
|
if not isinstance(generator, (DefaultGenerator, FilesGenerator)):
|
|
raise TypeError("generator must be DefaultGenerator or FilesGenerator")
|
|
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,
|
|
component_variant=generator.component_variant,
|
|
header_buttons=header_buttons,
|
|
)
|
|
return generator.generate(compiler, Path(build_directory).expanduser().resolve())
|
|
|
|
|
|
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,
|
|
component_variant: VariantSelection = 1,
|
|
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,
|
|
component_variant=component_variant,
|
|
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"]
|