obsidian_client.py•7.9 kB
"""
Obsidian REST API Client
Handles all communication with the Obsidian Local REST API plugin.
"""
import os
import json
from typing import Dict, Any, List, Optional
import httpx
import frontmatter
class ObsidianAPIError(Exception):
"""Base exception for Obsidian API errors."""
pass
class ObsidianClient:
"""Client for interacting with Obsidian Local REST API."""
def __init__(self):
"""Initialize the Obsidian client with environment variables."""
self.api_key = os.getenv("OBSIDIAN_API_KEY", "")
self.host = os.getenv("OBSIDIAN_HOST", "127.0.0.1")
self.port = os.getenv("OBSIDIAN_PORT", "27124")
self.base_url = f"https://{self.host}:{self.port}"
if not self.api_key:
raise ValueError("OBSIDIAN_API_KEY environment variable is required")
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Use httpx with disabled SSL verification for local self-signed certs
self.client = httpx.AsyncClient(
verify=False,
timeout=30.0,
headers=self.headers
)
async def close(self):
"""Close the HTTP client."""
await self.client.aclose()
def _handle_error(self, response: httpx.Response, operation: str) -> None:
"""Handle HTTP errors with actionable messages."""
if response.status_code == 401:
raise ObsidianAPIError(
f"Authentication failed for {operation}. "
"Check that OBSIDIAN_API_KEY is correct and the REST API plugin is running."
)
elif response.status_code == 404:
raise ObsidianAPIError(
f"Resource not found for {operation}. "
"Verify the file path exists in your vault."
)
elif response.status_code == 400:
raise ObsidianAPIError(
f"Bad request for {operation}: {response.text}. "
"Check that all parameters are valid."
)
else:
raise ObsidianAPIError(
f"API error for {operation} (status {response.status_code}): {response.text}"
)
async def get(self, endpoint: str) -> Dict[str, Any]:
"""Execute GET request to Obsidian API."""
url = f"{self.base_url}{endpoint}"
try:
response = await self.client.get(url)
if response.status_code == 200:
return response.json() if response.text else {}
else:
self._handle_error(response, f"GET {endpoint}")
except httpx.RequestError as e:
raise ObsidianAPIError(
f"Connection error for GET {endpoint}: {str(e)}. "
"Ensure Obsidian is running with the Local REST API plugin enabled."
)
async def post(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Execute POST request to Obsidian API."""
url = f"{self.base_url}{endpoint}"
try:
response = await self.client.post(url, json=data or {})
if response.status_code in (200, 201):
return response.json() if response.text else {"success": True}
else:
self._handle_error(response, f"POST {endpoint}")
except httpx.RequestError as e:
raise ObsidianAPIError(
f"Connection error for POST {endpoint}: {str(e)}. "
"Ensure Obsidian is running with the Local REST API plugin enabled."
)
async def put(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Execute PUT request to Obsidian API."""
url = f"{self.base_url}{endpoint}"
try:
response = await self.client.put(url, json=data or {})
if response.status_code in (200, 201):
return response.json() if response.text else {"success": True}
else:
self._handle_error(response, f"PUT {endpoint}")
except httpx.RequestError as e:
raise ObsidianAPIError(
f"Connection error for PUT {endpoint}: {str(e)}. "
"Ensure Obsidian is running with the Local REST API plugin enabled."
)
async def delete(self, endpoint: str) -> Dict[str, Any]:
"""Execute DELETE request to Obsidian API."""
url = f"{self.base_url}{endpoint}"
try:
response = await self.client.delete(url)
if response.status_code in (200, 204):
return {"success": True, "message": "Resource deleted successfully"}
else:
self._handle_error(response, f"DELETE {endpoint}")
except httpx.RequestError as e:
raise ObsidianAPIError(
f"Connection error for DELETE {endpoint}: {str(e)}. "
"Ensure Obsidian is running with the Local REST API plugin enabled."
)
# High-level operations
async def read_file(self, filepath: str) -> str:
"""Read file content from the vault."""
result = await self.get(f"/vault/{filepath}")
return result.get("content", "")
async def write_file(self, filepath: str, content: str) -> Dict[str, Any]:
"""Write content to a file in the vault."""
return await self.put(f"/vault/{filepath}", {"content": content})
async def append_to_file(self, filepath: str, content: str) -> Dict[str, Any]:
"""Append content to an existing file or create new file."""
try:
existing_content = await self.read_file(filepath)
new_content = existing_content + "\n" + content if existing_content else content
return await self.write_file(filepath, new_content)
except ObsidianAPIError:
# File doesn't exist, create it
return await self.write_file(filepath, content)
def parse_frontmatter(self, content: str) -> tuple[Dict[str, Any], str]:
"""
Parse frontmatter from content.
Returns:
Tuple of (frontmatter_dict, body_content)
"""
try:
post = frontmatter.loads(content)
return post.metadata, post.content
except Exception:
# No frontmatter found
return {}, content
def serialize_with_frontmatter(self, metadata: Dict[str, Any], body: str) -> str:
"""Serialize content with frontmatter."""
post = frontmatter.Post(body, **metadata)
return frontmatter.dumps(post)
async def update_file_frontmatter(
self,
filepath: str,
updates: Dict[str, Any]
) -> Dict[str, Any]:
"""
Update frontmatter in a file without modifying body content.
Args:
filepath: Path to the file
updates: Dictionary of frontmatter fields to update
Returns:
Dictionary with success status and updated frontmatter
"""
content = await self.read_file(filepath)
current_fm, body = self.parse_frontmatter(content)
# Merge updates into current frontmatter
current_fm.update(updates)
# Serialize and write back
new_content = self.serialize_with_frontmatter(current_fm, body)
await self.write_file(filepath, new_content)
return {
"success": True,
"frontmatter": current_fm
}
async def get_file_frontmatter(self, filepath: str) -> Dict[str, Any]:
"""Extract frontmatter from a file."""
content = await self.read_file(filepath)
metadata, _ = self.parse_frontmatter(content)
return metadata