"""OpenAPI 3.1 documentation helpers for ObsiGate (#72). This module keeps the API-documentation concerns out of ``backend/main.py``: * :data:`TAGS_METADATA` — human descriptions and ordering for every tag. * :func:`enrich_openapi_schema` — post-processes the schema generated by FastAPI to assign tags, add servers/security/examples and document common error responses. * :func:`render_api_landing` — a self-contained ``/api`` landing page that reads ``/openapi.json`` at runtime and groups endpoints by tag, with direct links to the Swagger UI (``/docs``) and ReDoc (``/redoc``). """ from __future__ import annotations import re from typing import Any # --------------------------------------------------------------------------- # Tag metadata # --------------------------------------------------------------------------- TAGS_METADATA: list[dict[str, str]] = [ {"name": "System", "description": "Health checks, runtime configuration, diagnostics, dashboard stats and the SSE event stream."}, {"name": "Auth", "description": "Authentication, sessions, user profile, MFA (TOTP/WebAuthn) and API keys."}, {"name": "Files", "description": "Browse, read, create, rename, move, delete and download vault files."}, {"name": "PDF", "description": "PDF metadata and byte-range streaming for inline viewing."}, {"name": "Vaults", "description": "List, add and remove vaults; per-vault display settings; attachment indexing."}, {"name": "Search", "description": "Full-text and advanced search, tags, title suggestions and the link graph."}, {"name": "Bookmarks", "description": "Recently opened files, bookmarks and saved searches."}, {"name": "Backups", "description": "Automatic file backups, diffs, restore, compression and purge."}, {"name": "Export", "description": "Export notes or whole vaults to HTML, Markdown bundle or ePub."}, {"name": "Guide", "description": "Download the in-app user guide as Markdown or PDF (mirrors the help modal, FR/EN)."}, {"name": "AI", "description": "AI-powered editor actions, provider status and model discovery."}, {"name": "BooksLM", "description": "Directory-scoped AI chat (NotebookLM-style) over a vault folder."}, {"name": "MCP", "description": "Model Context Protocol server (Streamable HTTP) exposing the shared AI tool layer to external clients (Claude Desktop, Cursor…)."}, {"name": "Collaboration", "description": "Real-time collaborative editing over WebSocket (`/ws/collab/{vault}/{path}`): Yjs/CRDT updates, awareness (cursors) and debounced server-side persistence."}, {"name": "Sharing", "description": "Create and manage public read-only share links for documents."}, {"name": "Webhooks", "description": "HTTP callbacks signed with HMAC-SHA256 for file events."}, {"name": "Conflicts", "description": "Detect and resolve Syncthing sync-conflict files."}, {"name": "Admin", "description": "Admin-only system monitoring: stats, audit log, backup stats and live stream."}, {"name": "Plugins", "description": "Install, enable and manage user plugins."}, {"name": "Push", "description": "Web Push (VAPID) subscription management and test notifications."}, {"name": "Frontend", "description": "Static assets and SPA fallback routes."}, ] API_DESCRIPTION = """ **ObsiGate** exposes a REST API covering the entire application: vaults, files, full-text search, backups, exports, AI actions, sharing and administration. ### Authentication When authentication is enabled, send a bearer token obtained from `POST /api/auth/login`: ``` Authorization: Bearer ``` The same token is also accepted as an HTTP-only cookie, so browser clients can simply use `credentials: "include"`. ### Real-time collaboration Besides the REST API, a WebSocket endpoint `GET /ws/collab/{vault}/{path}` (upgrade) powers simultaneous editing of the same file: clients exchange Yjs/CRDT updates and awareness (remote cursors), and the server persists the document 2 s after the last change. Authentication uses the `access_token` cookie (or a `token` query parameter) and vault access is enforced per connection. See the collaboration feature documentation (`docs/features/collaboration.md`) for the protocol. ### Interactive documentation * **Swagger UI** — [/docs](/docs): try requests directly from the browser. * **ReDoc** — [/redoc](/redoc): clean, reading-oriented reference. * **OpenAPI JSON** — [/openapi.json](/openapi.json): machine-readable schema. ### Errors Errors use the standard FastAPI envelope `{"detail": "..."}` with the appropriate HTTP status (`400`, `401`, `403`, `404`, `409`, `422`, `500`). """ # --------------------------------------------------------------------------- # Automatic tag assignment # --------------------------------------------------------------------------- # Rules are evaluated in order; the first match wins. Keep the most specific # prefixes first (e.g. BooksLM before AI, config/ai-* before config). _TAG_RULES: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"^/api/auth"), "Auth"), (re.compile(r"^/api/admin"), "Admin"), (re.compile(r"^/api/plugins"), "Plugins"), (re.compile(r"^/api/push"), "Push"), (re.compile(r"^/api/ai/bookslm"), "BooksLM"), (re.compile(r"^/api/ai"), "AI"), (re.compile(r"^/mcp"), "MCP"), (re.compile(r"^/api/config/ai-"), "AI"), (re.compile(r"^/api/share"), "Sharing"), (re.compile(r"^/api/shares"), "Sharing"), (re.compile(r"^/s/"), "Sharing"), (re.compile(r"^/api/webhooks"), "Webhooks"), (re.compile(r"^/api/conflicts"), "Conflicts"), (re.compile(r"^/api/backups"), "Backups"), (re.compile(r"^/api/file/[^/]+/(backups|diff|restore)"), "Backups"), (re.compile(r"^/api/export"), "Export"), (re.compile(r"^/api/guide"), "Guide"), (re.compile(r"^/api/file/[^/]+/pdf"), "PDF"), (re.compile(r"^/api/search"), "Search"), (re.compile(r"^/api/tags"), "Search"), (re.compile(r"^/api/tree-search"), "Search"), (re.compile(r"^/api/suggest"), "Search"), (re.compile(r"^/api/graph"), "Search"), (re.compile(r"^/api/recent"), "Bookmarks"), (re.compile(r"^/api/bookmarks"), "Bookmarks"), (re.compile(r"^/api/saved-searches"), "Bookmarks"), (re.compile(r"^/api/vaults"), "Vaults"), (re.compile(r"^/api/vault/"), "Vaults"), (re.compile(r"^/api/index/reload"), "Vaults"), (re.compile(r"^/api/attachments"), "Vaults"), (re.compile(r"^/api/browse"), "Files"), (re.compile(r"^/api/file"), "Files"), (re.compile(r"^/api/directory"), "Files"), (re.compile(r"^/api/move"), "Files"), (re.compile(r"^/api/image"), "Files"), (re.compile(r"^/api/health"), "System"), (re.compile(r"^/api/config"), "System"), (re.compile(r"^/api/diagnostics"), "System"), (re.compile(r"^/api/dashboard"), "System"), (re.compile(r"^/api/events"), "System"), ] def tag_for_path(path: str) -> str: """Return the documentation tag for an API path (``Frontend`` as fallback).""" for pattern, tag in _TAG_RULES: if pattern.match(path): return tag return "Frontend" # Normalise tags declared by individual routers (``tags=["auth"]``) to the # canonical, documented tag names. _TAG_ALIASES: dict[str, str] = { "auth": "Auth", "admin": "Admin", "plugins": "Plugins", "push": "Push", "files": "Files", "vaults": "Vaults", "search": "Search", "backups": "Backups", "export": "Export", "sharing": "Sharing", "webhooks": "Webhooks", "conflicts": "Conflicts", "system": "System", "frontend": "Frontend", "ai": "AI", "bookslm": "BooksLM", "mcp": "MCP", "pdf": "PDF", "bookmarks": "Bookmarks", } _CANONICAL_TAGS = {tag["name"] for tag in TAGS_METADATA} def canonical_tag(name: str) -> str: """Map a router-declared tag to its documented name (unchanged if unknown).""" mapped = _TAG_ALIASES.get(name.lower(), name) return mapped if mapped in _CANONICAL_TAGS else name # --------------------------------------------------------------------------- # Request/response examples injected into the schema # --------------------------------------------------------------------------- _ENDPOINT_EXAMPLES: dict[tuple[str, str], dict[str, Any]] = { ("post", "/api/file/{vault_name}"): { "request": {"path": "notes/Nouvelle note.md", "content": "# Titre\n\nContenu"}, "response": {"success": True, "path": "notes/Nouvelle note.md"}, }, ("put", "/api/file/{vault_name}/save"): { "request": {"path": "notes/Accueil.md", "content": "# Accueil\n\nMis à jour."}, "response": {"status": "ok", "vault": "TestVault", "path": "notes/Accueil.md", "size": 26}, }, ("put", "/api/file/{vault_name}/xlsx/save"): { "request": {"sheet": "Budget", "cells": {"B1": "250"}, "allow_formula": False, "force": False}, "response": {"status": "ok", "vault": "TestVault", "path": "data/budget.xlsx", "size": 1}, }, # GET : pas d'exemple de requête (un requestBody sur un GET serait un OpenAPI # invalide) — les paramètres sont documentés par leurs Query(). ("get", "/api/file/{vault_name}/xlsx/sheet"): { "response": { "vault": "TestVault", "path": "data/budget.xlsx", "sheet": "Budget", "offset": 0, "limit": 200, "rows": 2, "cols": 2, "total_rows": 640, "total_cols": 12, "max_rows": 500, "max_cols": 40, "truncated": True, "has_more": True, "html": "…
", }, }, ("post", "/api/search/replace"): { "request": {"query": "Python", "replacement": "Python 3", "vault": "all", "dry_run": True}, "response": {"matches": [{"vault": "TestVault", "path": "note1.md", "title": "Python", "match_count": 3}], "total_matches": 3, "dry_run": True}, }, ("post", "/api/ai/improve"): { "request": {"text": "ce texte est bof", "provider": "deepseek"}, "response": {"result": "Ce texte mérite d'être amélioré.", "provider": "deepseek"}, }, ("post", "/api/ai/bookslm/chat"): { "request": {"vault": "TestVault", "directory": "projets", "message": "Résume ce dossier", "conversation_history": []}, "response": {"token": "Voici un résumé…", "provider": "deepseek", "model": "deepseek-chat"}, }, ("post", "/api/share/{vault_name}"): { "request": {"path": "notes/Accueil.md", "expires_in_hours": 168}, "response": {"id": "abc…", "token": "abc…", "vault": "TestVault", "path": "notes/Accueil.md", "url": "/s/abc…"}, }, ("post", "/api/webhooks"): { "request": {"name": "CI", "url": "https://example.com/hook", "events": ["file_modified"], "secret": "s3cr3t"}, "response": {"id": "3f2c…", "name": "CI", "url": "https://example.com/hook", "events": ["file_modified"], "enabled": True}, }, } # Common error responses documented on every operation. _COMMON_ERRORS: dict[str, dict[str, Any]] = { "401": { "description": "Authentication required or token expired", "content": {"application/json": {"example": {"detail": "Not authenticated"}}}, }, "403": { "description": "Access to the vault is denied", "content": {"application/json": {"example": {"detail": "Accès refusé à la vault 'Notes'"}}}, }, "404": { "description": "Resource not found", "content": {"application/json": {"example": {"detail": "File not found: notes/x.md"}}}, }, "422": { "description": "Validation error", "content": {"application/json": {"example": {"detail": [{"loc": ["query", "path"], "msg": "field required", "type": "value_error.missing"}]}}}, }, "500": { "description": "Internal server error", "content": {"application/json": {"example": {"detail": "Internal server error"}}}, }, } # Paths that return a binary/streaming body — never add a JSON error example # to their 200 response, and skip the 500 JSON example for streams. _BINARY_MEDIA = ("application/pdf", "application/octet-stream", "image/", "text/event-stream") def _is_binary_operation(operation: dict[str, Any]) -> bool: content = operation.get("responses", {}).get("200", {}).get("content", {}) return any(media.startswith(_BINARY_MEDIA) for media in content) # --------------------------------------------------------------------------- # MCP endpoint (not a FastAPI route: custom ASGI mount) — documented manually # --------------------------------------------------------------------------- _MCP_DESCRIPTION = ( "**Model Context Protocol** server over Streamable HTTP (JSON-RPC 2.0). " "Exposes the shared AI tool layer to external MCP clients (Claude Desktop, " "Cursor…). Authentication uses `Authorization: Bearer ` (the same " "token as the REST API).\n\n" "Primitives: read/search **tools** directly; write/destructive tools as a " "two-step `propose_` / `apply_` pair (signed, single-use " "confirmation token); **resources** `vault://` and " "`vault:///` (read-only, secrets redacted); **prompts** " "`summarize-directory`, `generate-note`, `find-related`.\n\n" "See `docs/MCP_GUIDE.md` for client setup." ) def _inject_mcp_path(schema: dict[str, Any]) -> None: """Add the MCP Streamable HTTP endpoint to the schema (idempotent).""" paths = schema.setdefault("paths", {}) if "/mcp" in paths: return paths["/mcp"] = { "post": { "tags": ["MCP"], "summary": "MCP Streamable HTTP endpoint (JSON-RPC 2.0)", "operationId": "mcp_streamable_http", "description": _MCP_DESCRIPTION, "requestBody": { "required": True, "content": { "application/json": { "example": { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}, } } }, }, "responses": { "200": { "description": "JSON-RPC response (or 202 for notifications)", "content": { "application/json": { "example": { "jsonrpc": "2.0", "id": 1, "result": {"tools": []}, } } }, } }, "security": [{"bearerAuth": []}], } } def enrich_openapi_schema(schema: dict[str, Any]) -> dict[str, Any]: """Enrich a FastAPI-generated OpenAPI schema in place and return it. The transformation is idempotent: calling it twice yields the same result. """ schema.setdefault("openapi", "3.1.0") info = schema.setdefault("info", {}) info.setdefault("description", API_DESCRIPTION.strip()) info.setdefault("contact", {"name": "ObsiGate", "url": "https://git.dracodev.net/Projets/ObsiGate"}) info.setdefault("license", {"name": "MIT"}) schema.setdefault("servers", [{"url": "/", "description": "Current ObsiGate instance"}]) schema["externalDocs"] = { "description": "ObsiGate source repository", "url": "https://git.dracodev.net/Projets/ObsiGate", } schema["tags"] = TAGS_METADATA _inject_mcp_path(schema) components = schema.setdefault("components", {}) security_schemes = components.setdefault("securitySchemes", {}) security_schemes.setdefault("bearerAuth", { "type": "http", "scheme": "bearer", "bearerFormat": "JWT", "description": "JWT access token issued by POST /api/auth/login.", }) security_schemes.setdefault("cookieAuth", { "type": "apiKey", "in": "cookie", "name": "obsigate_token", "description": "HTTP-only session cookie (browser clients).", }) components.setdefault("schemas", {}).setdefault("ErrorResponse", { "type": "object", "properties": {"detail": {"title": "Detail", "description": "Error message"}}, }) for path, operations in schema.get("paths", {}).items(): default_tag = tag_for_path(path) for method, operation in operations.items(): if not isinstance(operation, dict): continue if operation.get("tags"): operation["tags"] = [canonical_tag(t) for t in operation["tags"]] else: operation["tags"] = [default_tag] # Document common error responses without overriding explicit ones. responses = operation.setdefault("responses", {}) binary = _is_binary_operation(operation) for code, spec in _COMMON_ERRORS.items(): if code == "500" and binary: continue responses.setdefault(code, spec) # Inject request/response examples for key endpoints. example = _ENDPOINT_EXAMPLES.get((method, path)) if example: if "request" in example: body = operation.setdefault("requestBody", {}) content = body.setdefault("content", {}).setdefault("application/json", {}) content.setdefault("example", example["request"]) if "response" in example: ok = responses.setdefault("200", {}) content = ok.setdefault("content", {}).setdefault("application/json", {}) content.setdefault("example", example["response"]) return schema # --------------------------------------------------------------------------- # /api landing page # --------------------------------------------------------------------------- _METHOD_COLORS = { "get": "#2563eb", "post": "#16a34a", "put": "#d97706", "patch": "#7c3aed", "delete": "#dc2626", "head": "#64748b", "options": "#64748b", } def render_api_landing(version: str = "") -> str: """Return the self-contained HTML for the ``/api`` documentation page.""" return f""" ObsiGate API{(' — v' + version) if version else ''}

ObsiGate API

Interactive reference for the ObsiGate REST API (OpenAPI 3.1).

Swagger UI — try it ReDoc — reference OpenAPI JSON ← Back to ObsiGate
Loading specification…

Loading endpoints…

Generated from the live OpenAPI schema. Authentication: Authorization: Bearer <token>.
""" def _js_colors() -> str: """Serialise the method→colour map for the landing page script.""" import json return json.dumps(_METHOD_COLORS)