0.3.1: sync locking, dry-run API, env var interpolation

- Per-KB asyncio.Lock prevents overlapping syncs from scheduler,
  webhook, and API triggers running simultaneously
- POST /sync/{id}?dry_run=true previews changes without uploading
- ${VAR} and ${VAR:-default} interpolation in all .oikb.yaml strings
  for GitOps workflows where secrets come from the runtime

Author: tjbck

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [0.3.1] - 2025-05-21
+
+### Added
+
+- Per-KB sync locking — prevents overlapping syncs to the same Knowledge Base. If a webhook fires while a scheduled sync is running, the duplicate is skipped with a log message.
+- Dry-run via API — `POST /sync/{id}?dry_run=true` previews changes without uploading, returns added/modified/deleted counts.
+- Environment variable interpolation in `.oikb.yaml` — `${VAR}` and `${VAR:-default}` syntax in all string values. Enables GitOps workflows where secrets come from the runtime.
+
 ## [0.3.0] - 2025-05-21
 
 ### Added

README.md

@@ -170,6 +170,18 @@ Resolved in order (highest priority wins):
 2. **Environment variables** (`OPEN_WEBUI_URL`, `OPEN_WEBUI_API_KEY`)
 3. **Config file** (`~/.config/oikb/config.yaml`)
 
+All string values in `.oikb.yaml` support `${VAR}` and `${VAR:-default}` interpolation:
+
+```yaml
+sources:
+  - name: docs
+    source: github:${GITHUB_ORG}/docs
+    kb-id: ${KB_DOCS_ID}
+    token: ${GITHUB_TOKEN}
+    notify:
+      url: ${SLACK_WEBHOOK:-https://hooks.slack.com/default}
+```
+
 ## History
 
 ```bash

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oikb"
-version = "0.3.0"
+version = "0.3.1"
 description = "Sync anything to Open WebUI Knowledge Bases"
 readme = "README.md"
 authors = [

src/oikb/__init__.py

@@ -1,3 +1,3 @@
 """oikb — Open WebUI Knowledge Base CLI."""
 
-__version__ = "0.3.0"
+__version__ = "0.3.1"

src/oikb/cli.py

@@ -5,6 +5,7 @@
 import os
 import sys
 from pathlib import Path
+from typing import Any
 
 import click
 
@@ -272,6 +273,28 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
     return FilesystemConnector(source)
 
 
+def _interpolate_env(obj: Any) -> Any:
+    """Recursively interpolate ${VAR} and ${VAR:-default} in string values."""
+    import re
+
+    _ENV_RE = re.compile(r"\$\{([^}]+)\}")
+
+    def _replace(match: re.Match) -> str:
+        expr = match.group(1)
+        if ":-" in expr:
+            var, default = expr.split(":-", 1)
+            return os.environ.get(var, default)
+        return os.environ.get(expr, match.group(0))
+
+    if isinstance(obj, str):
+        return _ENV_RE.sub(_replace, obj)
+    if isinstance(obj, dict):
+        return {k: _interpolate_env(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_interpolate_env(item) for item in obj]
+    return obj
+
+
 def _load_oikb_yaml() -> list[dict] | None:
     """Load .oikb.yaml from the current directory if it exists."""
     import yaml
@@ -286,6 +309,9 @@ def _load_oikb_yaml() -> list[dict] | None:
     if not data:
         return None
 
+    # Interpolate environment variables in all string values.
+    data = _interpolate_env(data)
+
     # Prefer sources: (new), fall back to sync: (legacy).
     entries = data.get("sources") or data.get("sync")
     return entries or None
@@ -789,6 +815,7 @@ def daemon(port: int, no_server: bool, config_file: str | None, log_format: str
         import yaml
         with open(config_file) as f:
             data = yaml.safe_load(f)
+        data = _interpolate_env(data) if data else data
         entries = (data.get("sources") or data.get("sync", [])) if data else []
     else:
         entries = _load_oikb_yaml()

src/oikb/daemon.py

@@ -40,11 +40,12 @@ async def verify_api_key(
 app = FastAPI(
     title="oikb",
     description="Sync engine for Open WebUI Knowledge Bases. Trigger syncs, check status, and query history.",
-    version="0.3.0",
+    version="0.3.1",
 )
 
 # Runtime state populated by start_daemon().
 _scheduler_state: dict[str, dict[str, Any]] = {}
+_sync_locks: dict[str, asyncio.Lock] = {}
 _history: SyncHistory | None = None
 _entries: list[dict] = []
 _shutdown_event: asyncio.Event | None = None
@@ -128,10 +129,13 @@ async def history_endpoint(
     summary="Trigger an immediate sync by alias or KB ID",
     dependencies=[Depends(verify_api_key)],
 )
-async def trigger_sync(identifier: str):
-    """Triggers an immediate sync matching the given alias or Knowledge Base ID. The sync runs asynchronously in the background. Use get_sync_status to check progress."""
+async def trigger_sync(identifier: str, dry_run: bool = False):
+    """Triggers an immediate sync matching the given alias or Knowledge Base ID. The sync runs asynchronously in the background. Use get_sync_status to check progress. Set dry_run=true to preview changes without uploading."""
     for entry in _entries:
         if entry.get("name") == identifier or entry.get("kb-id") == identifier:
+            if dry_run:
+                result = await _run_entry(entry, dry_run=True)
+                return {"dry_run": True, "name": entry.get("name"), "kb_id": entry.get("kb-id"), "result": result}
             asyncio.create_task(_run_entry(entry))
             return {"triggered": True, "name": entry.get("name"), "kb_id": entry.get("kb-id")}
     return {"triggered": False, "error": f"No entry matching '{identifier}'"}
@@ -179,8 +183,29 @@ async def _send_notification(entry: dict, payload: dict) -> None:
         log.warning(f"Notification failed for {source}: {exc}")
 
 
-async def _run_entry(entry: dict) -> None:
-    """Run a single sync for an entry."""
+async def _run_entry(entry: dict, dry_run: bool = False) -> dict | None:
+    """Run a single sync for an entry.
+
+    Uses a per-KB lock to prevent overlapping syncs to the same
+    Knowledge Base (e.g. webhook fires while a scheduled sync is running).
+    """
+    kb_id = entry["kb-id"]
+
+    # Get or create a lock for this KB.
+    if kb_id not in _sync_locks:
+        _sync_locks[kb_id] = asyncio.Lock()
+    lock = _sync_locks[kb_id]
+
+    if lock.locked():
+        log.info(f"Skipping {entry.get('source', '?')} — sync already running for {kb_id}")
+        return {"skipped": True, "reason": "sync already running"} if dry_run else None
+
+    async with lock:
+        return await _run_entry_locked(entry, dry_run=dry_run)
+
+
+async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
+    """Inner sync logic, called under the per-KB lock."""
     from oikb.cli import _make_client, _resolve_connector
     from oikb.sync import run_sync
 
@@ -218,16 +243,27 @@ async def _run_entry(entry: dict) -> None:
                 max_size=parse_size(ms),
             )
 
-        result = run_sync(
+        result = await asyncio.to_thread(
+            run_sync,
             client=client,
             connector=connector,
             kb_id=kb_id,
+            dry_run=dry_run,
             quiet=True,
             manifest_filter=mf,
             concurrency=entry.get("concurrency", 1),
         )
         client.close()
 
+        if dry_run:
+            return {
+                "added": result.added,
+                "modified": result.modified,
+                "deleted": result.deleted,
+                "unmodified": result.unmodified,
+                "summary": result.summary(),
+            }
+
         duration_s = time.time() - started_at
         duration_ms = int(duration_s * 1000)
         status = "success" if not result.errors else "partial"