"""Docmost API Client - Based on actual API endpoints."""
import httpx
from typing import Any
from pydantic import BaseModel
class DocmostConfig(BaseModel):
"""Configuration for Docmost API."""
base_url: str
auth_token: str
timeout: float = 30.0
class DocmostClient:
"""HTTP client for Docmost API."""
def __init__(self, config: DocmostConfig):
self.config = config
self.base_url = config.base_url.rstrip("/")
self._client = httpx.AsyncClient(
base_url=self.base_url,
timeout=config.timeout,
headers={
"Content-Type": "application/json",
"Accept": "application/json",
},
cookies={"authToken": config.auth_token},
)
async def close(self):
"""Close the HTTP client."""
await self._client.aclose()
async def _request(
self,
method: str,
endpoint: str,
json: dict | None = None,
) -> dict[str, Any]:
"""Make an API request."""
response = await self._client.request(
method=method,
url=f"/api/{endpoint}",
json=json if json is not None else {},
)
response.raise_for_status()
return response.json()
# ==================== Pages ====================
async def get_page_info(self, page_id: str) -> dict[str, Any]:
"""Get page information by ID or slug."""
return await self._request("POST", "pages/info", json={"pageId": page_id})
async def get_sidebar_pages(
self,
space_id: str,
page_id: str | None = None,
page: int = 1,
) -> dict[str, Any]:
"""Get sidebar pages for a space, optionally filtered by parent page."""
payload = {"spaceId": space_id, "page": page}
if page_id:
payload["pageId"] = page_id
return await self._request("POST", "pages/sidebar-pages", json=payload)
async def get_recent_pages(self, space_id: str) -> dict[str, Any]:
"""Get recently updated pages in a space."""
return await self._request("POST", "pages/recent", json={"spaceId": space_id})
async def create_page(
self,
title: str,
space_id: str,
parent_page_id: str | None = None,
content: dict | None = None,
) -> dict[str, Any]:
"""Create a new page."""
payload = {"title": title, "spaceId": space_id}
if parent_page_id:
payload["parentPageId"] = parent_page_id
if content:
payload["content"] = content
return await self._request("POST", "pages/create", json=payload)
async def update_page(
self,
page_id: str,
title: str | None = None,
content: dict | None = None,
) -> dict[str, Any]:
"""Update an existing page."""
payload = {"pageId": page_id}
if title:
payload["title"] = title
if content:
payload["content"] = content
return await self._request("POST", "pages/update", json=payload)
async def delete_page(self, page_id: str) -> dict[str, Any]:
"""Delete a page."""
return await self._request("POST", "pages/delete", json={"pageId": page_id})
# ==================== Spaces ====================
async def list_spaces(self, limit: int = 20, page: int = 1) -> dict[str, Any]:
"""List all spaces in the workspace."""
return await self._request("POST", "spaces", json={"limit": limit, "page": page})
async def get_space_info(self, space_id: str) -> dict[str, Any]:
"""Get space information by ID or slug."""
return await self._request("POST", "spaces/info", json={"spaceId": space_id})
# ==================== Users ====================
async def get_current_user(self) -> dict[str, Any]:
"""Get current authenticated user info."""
return await self._request("POST", "users/me")
async def list_workspace_members(
self, limit: int = 50, page: int = 1
) -> dict[str, Any]:
"""List workspace members."""
return await self._request(
"POST", "workspaces/members", json={"limit": limit, "page": page}
)
# ==================== Shares ====================
async def get_page_shares(self, page_id: str) -> dict[str, Any]:
"""Get share settings for a page."""
return await self._request("POST", "shares/for-page", json={"pageId": page_id})
# ==================== Comments ====================
async def list_comments(
self, page_id: str, limit: int = 50
) -> dict[str, Any]:
"""List comments on a page."""
return await self._request(
"POST", "comments", json={"pageId": page_id, "limit": limit}
)
async def create_comment(
self, page_id: str, content: str, parent_comment_id: str | None = None
) -> dict[str, Any]:
"""Create a comment on a page."""
payload = {"pageId": page_id, "content": content}
if parent_comment_id:
payload["parentCommentId"] = parent_comment_id
return await self._request("POST", "comments/create", json=payload)
# ==================== Page History ====================
async def get_page_history(
self, page_id: str, limit: int = 20
) -> dict[str, Any]:
"""Get page revision history."""
return await self._request(
"POST", "pages/history", json={"pageId": page_id, "limit": limit}
)
