Integrating Obsidian with Astro

July 17, 2025 in astro and obsidian.

How to integrate Obsidian with Astro?

If you write in Obsidian and build your blog with Astro, there’s a neat way to connect the two — and it’s surprisingly simple.

With the help of the obsidian-livesync plugin and a small Python script, you can turn your Obsidian notes into live blog posts that automatically sync to your website. Here’s how it works.

Tools You’ll Need

  • Obsidian with the obsidian-livesync plugin
  • A CouchDB instance (local or remote)
  • A Python script to fetch notes from CouchDB and save them locally
  • An Astro project that reads Markdown files from a folder
  • Optionally: a cron job or file watcher to automate the sync

How It Works

  1. You write your content in Obsidian, like usual.
  2. The obsidian-livesync plugin syncs your vault to CouchDB.
  3. A Python script pulls documents from CouchDB and writes them to your Astro src/content directory.
  4. Astro rebuilds the site — or you trigger a rebuild through a deploy or GitHub Action.

The Python Script

Here’s a basic version that fetches your notes using environment variables:

import http.client
import urllib.parse
import base64
import json
import os
import shutil
import sys

# --- Configuration from environment ---
COUCHDB_HOST = os.getenv("COUCHDB_HOST")
DB_NAME = os.getenv("DB_NAME")
COUCHDB_USER = os.getenv("COUCHDB_USER")
COUCHDB_PASSWORD = os.getenv("COUCHDB_PASSWORD")
OUTPUT_DIR = os.getenv("OUTPUT_DIR", "src/data/blog")

# Validate required env vars
missing = [
    var
    for var in ["COUCHDB_HOST", "DB_NAME", "COUCHDB_USER", "COUCHDB_PASSWORD"]
    if not os.getenv(var)
]
if missing:
    sys.exit(f"❌ Missing environment variables: {', '.join(missing)}")

AUTH_HEADER = (
    "Basic " + base64.b64encode(f"{COUCHDB_USER}:{COUCHDB_PASSWORD}".encode()).decode()
)

# --- Functions ---


def http_get(path, params=None):
    if params:
        query = urllib.parse.urlencode(params)
        full_path = f"{path}?{query}"
    else:
        full_path = path

    conn = http.client.HTTPConnection(COUCHDB_HOST)
    headers = {
        "Authorization": AUTH_HEADER,
        "Accept": "application/json",
    }
    conn.request("GET", full_path, headers=headers)
    response = conn.getresponse()
    data = response.read().decode()
    conn.close()

    if response.status != 200:
        raise Exception(
            f"HTTP GET {full_path} failed: {response.status} {response.reason} - {data}"
        )

    return json.loads(data)


def fetch_doc(doc_id):
    encoded_id = urllib.parse.quote(doc_id, safe="")
    path = f"/{DB_NAME}/{encoded_id}"
    return http_get(path)


def get_all_doc_ids():
    path = f"/{DB_NAME}/_all_docs"
    resp = http_get(path, params={"include_docs": "false"})
    rows = resp.get("rows", [])
    return [
        row["id"]
        for row in rows
        if row["id"].startswith("website/") and row["id"].endswith(".md")
    ]


def get_full_markdown(doc_id):
    meta = fetch_doc(doc_id)
    if meta.get("deleted", False):
        return None

    children = meta.get("children", [])
    contents = []

    for child_id in children:
        block = fetch_doc(child_id)
        contents.append(block.get("data", ""))

    return "".join(contents)


def save_markdown(doc_id, content):
    if doc_id.startswith("website/"):
        rel_path = doc_id[len("website/") :]
    else:
        rel_path = doc_id
    full_path = os.path.join(OUTPUT_DIR, rel_path)
    os.makedirs(os.path.dirname(full_path), exist_ok=True)

    with open(full_path, "w", encoding="utf-8") as f:
        f.write(content)

    print(f"✅ Saved: {full_path}")


def clear_output_dir():
    if os.path.exists(OUTPUT_DIR):
        shutil.rmtree(OUTPUT_DIR)
    os.makedirs(OUTPUT_DIR, exist_ok=True)


# --- Main ---

if __name__ == "__main__":
    try:
        clear_output_dir()
        doc_ids = get_all_doc_ids()

        for doc_id in doc_ids:
            try:
                content = get_full_markdown(doc_id)
                if content:
                    save_markdown(doc_id, content)
            except Exception as e:
                print(f"⚠️ Failed to process {doc_id}: {e}")

        print("🎉 All markdown files exported.")
    except Exception as e:
        print(f"❌ Error: {e}")

This script grabs all documents under website/ and writes them as Markdown files into Astro’s blog directory.

Automating It

You can run the script manually, or automate it:

  • Use a cron job (crontab -e) to run it every few minutes
  • Or set it up as a GitHub Action on a schedule or push
  • Or use a live-server or Astro’s dev mode with file watching

Whatever you pick, your blog updates every time you change a note.