122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""Thin async wrapper around the BookStack REST API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from html import unescape
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import Settings
|
|
|
|
|
|
class BookStackError(RuntimeError):
|
|
"""Raised when BookStack returns an error response."""
|
|
|
|
|
|
_TAG_RE = re.compile(r"<[^>]+>")
|
|
|
|
|
|
def strip_html(value: str | None) -> str:
|
|
"""Turn BookStack's preview HTML into plain text."""
|
|
if not value:
|
|
return ""
|
|
text = _TAG_RE.sub(" ", value)
|
|
return re.sub(r"\s+", " ", unescape(text)).strip()
|
|
|
|
|
|
class BookStackClient:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._settings = settings
|
|
self._client = httpx.AsyncClient(
|
|
base_url=settings.api_base,
|
|
timeout=settings.request_timeout,
|
|
verify=settings.bookstack_verify_tls,
|
|
headers={
|
|
"Authorization": (
|
|
f"Token {settings.bookstack_token_id}:{settings.bookstack_token_secret}"
|
|
),
|
|
"Accept": "application/json",
|
|
"User-Agent": "bookstack-mcp/1.0",
|
|
},
|
|
follow_redirects=True,
|
|
)
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|
|
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
json: dict[str, Any] | None = None,
|
|
raw: bool = False,
|
|
) -> Any:
|
|
try:
|
|
response = await self._client.request(
|
|
method, path, params=params, json=json
|
|
)
|
|
except httpx.HTTPError as exc: # network/DNS/TLS failures
|
|
raise BookStackError(f"Could not reach BookStack at {self._settings.bookstack_url}: {exc}") from exc
|
|
|
|
if response.status_code == 401:
|
|
raise BookStackError(
|
|
"BookStack rejected the API token (401). Check BOOKSTACK_TOKEN_ID / "
|
|
"BOOKSTACK_TOKEN_SECRET and that the token's user has the "
|
|
"'Access System API' permission."
|
|
)
|
|
if response.status_code == 403:
|
|
raise BookStackError(
|
|
"BookStack denied access (403). The token's user lacks permission for "
|
|
"this content or action."
|
|
)
|
|
if response.status_code == 404:
|
|
raise BookStackError("Not found in BookStack (404).")
|
|
if response.status_code == 429:
|
|
raise BookStackError(
|
|
"BookStack rate limit hit (429). Default is 180 requests/minute per user."
|
|
)
|
|
if response.status_code >= 400:
|
|
detail = response.text[:500]
|
|
raise BookStackError(f"BookStack returned HTTP {response.status_code}: {detail}")
|
|
|
|
if raw:
|
|
return response.text
|
|
if not response.content:
|
|
return {}
|
|
return response.json()
|
|
|
|
# ---------- listing helpers ----------
|
|
|
|
async def list_endpoint(
|
|
self,
|
|
endpoint: str,
|
|
*,
|
|
count: int = 25,
|
|
offset: int = 0,
|
|
sort: str | None = None,
|
|
filters: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
params: dict[str, Any] = {"count": min(max(count, 1), 500), "offset": max(offset, 0)}
|
|
if sort:
|
|
params["sort"] = sort
|
|
for key, value in (filters or {}).items():
|
|
params[f"filter[{key}]"] = value
|
|
return await self.request("GET", endpoint, params=params)
|
|
|
|
async def search(self, query: str, *, count: int, page: int) -> dict[str, Any]:
|
|
return await self.request(
|
|
"GET",
|
|
"/search",
|
|
params={"query": query, "count": min(max(count, 1), 100), "page": max(page, 1)},
|
|
)
|
|
|
|
async def export(self, kind: str, item_id: int, fmt: str) -> str:
|
|
return await self.request("GET", f"/{kind}/{item_id}/export/{fmt}", raw=True)
|
|
|
|
def web_url(self, path: str) -> str:
|
|
return f"{self._settings.bookstack_url}/{path.lstrip('/')}"
|