obsidian-sync

a short description
git clone https://git.skysyrup.dev/obsidian-sync.git
Log | Files | Refs | README

api.py (4428B)

      1 """
      2 Obsidian Sync REST API client.
      3 
      4 Reversed from Obsidian 1.11.7 app.js.
      5 
      6 Base URL: https://api.obsidian.md  (SZ = "https://" + ["api","obsidian","md"].join("."))
      7 All POST requests send JSON and receive JSON.
      8 
      9 Endpoints:
     10   POST /user/signin          {email, password, mfa?}        -> {token, email, name, license}
     11   POST /user/signout         {token}                         -> {}
     12   POST /user/info            {token}                         -> {email, name, license}
     13   POST /user/authtoken       {token}                         -> {token, email, name, license}
     14   POST /vault/list           {token, supported_encryption_version: 3} -> {vaults: [...], shared: [...]}
     15   POST /vault/create         {token, name, keyhash, salt, region, encryption_version} -> vault
     16   POST /vault/access         {token, vault_uid, keyhash, host, encryption_version}    -> {}
     17   POST /vault/regions        {token, host}                   -> {regions: [...]}
     18   POST /subscription/list    {token}                         -> {subscriptions: [...]}
     19 
     20 Vault object fields:
     21   id, name, salt, host, encryption_version, keyhash, ...
     22 """
     23 
     24 from __future__ import annotations
     25 
     26 import asyncio
     27 import aiohttp
     28 from dataclasses import dataclass, field
     29 from typing import Optional, Any
     30 
     31 API_BASE = "https://api.obsidian.md"
     32 
     33 
     34 @dataclass
     35 class Vault:
     36     id: str
     37     name: str
     38     salt: str
     39     host: str
     40     encryption_version: int
     41     keyhash: str = ""
     42     password: str = ""  # Set by user, not from API
     43 
     44 
     45 @dataclass
     46 class UserSession:
     47     token: str
     48     email: str = ""
     49     name: str = ""
     50     license: str = ""
     51 
     52 
     53 class ObsidianAPIError(Exception):
     54     def __init__(self, error: str, response: Optional[dict] = None):
     55         super().__init__(error)
     56         self.error = error
     57         self.response = response
     58 
     59 
     60 _DEFAULT_HEADERS = {
     61     "Content-Type": "application/json",
     62     "Referer": "https://obsidian.md/",
     63     "Origin": "https://obsidian.md",
     64 }
     65 
     66 
     67 async def _api_post(session: aiohttp.ClientSession, endpoint: str, payload: dict) -> dict:
     68     """Make an authenticated POST request to the Obsidian API."""
     69     url = API_BASE + endpoint
     70     async with session.post(url, json=payload, headers=_DEFAULT_HEADERS) as resp:
     71         data = await resp.json(content_type=None)
     72         if "error" in data:
     73             raise ObsidianAPIError(data["error"], data)
     74         return data
     75 
     76 
     77 async def signin(email: str, password: str, mfa: str = "") -> UserSession:
     78     """Sign in and return a session token."""
     79     async with aiohttp.ClientSession() as session:
     80         payload: dict[str, Any] = {"email": email, "password": password}
     81         if mfa:
     82             payload["mfa"] = mfa
     83         data = await _api_post(session, "/user/signin", payload)
     84         return UserSession(
     85             token=data.get("token", ""),
     86             email=data.get("email", ""),
     87             name=data.get("name", ""),
     88             license=data.get("license", ""),
     89         )
     90 
     91 
     92 async def signout(token: str) -> None:
     93     """Sign out."""
     94     async with aiohttp.ClientSession() as session:
     95         try:
     96             await _api_post(session, "/user/signout", {"token": token})
     97         except ObsidianAPIError as e:
     98             if "Not logged in" not in str(e):
     99                 raise
    100 
    101 
    102 async def get_user_info(token: str) -> dict:
    103     """Get current user info."""
    104     async with aiohttp.ClientSession() as session:
    105         return await _api_post(session, "/user/info", {"token": token})
    106 
    107 
    108 async def list_vaults(token: str) -> list[Vault]:
    109     """List all vaults for the current user."""
    110     async with aiohttp.ClientSession() as session:
    111         data = await _api_post(
    112             session, "/vault/list",
    113             {"token": token, "supported_encryption_version": 3}
    114         )
    115     vaults = []
    116     for v in data.get("vaults", []) + data.get("shared", []):
    117         if "encryption_version" not in v:
    118             v["encryption_version"] = 0
    119         vaults.append(Vault(
    120             id=v["id"],
    121             name=v["name"],
    122             salt=v.get("salt", ""),
    123             host=v.get("host", ""),
    124             encryption_version=v["encryption_version"],
    125             keyhash=v.get("keyhash", ""),
    126         ))
    127     return vaults
    128 
    129 
    130 async def list_regions(token: str, host: str = "") -> list[str]:
    131     """List available sync regions."""
    132     async with aiohttp.ClientSession() as session:
    133         data = await _api_post(session, "/vault/regions", {"token": token, "host": host})
    134         return data.get("regions", [])