From 6f8d0becd2de3ddc504bc8278bcb824123c2c8c6 Mon Sep 17 00:00:00 2001 From: Kalyan Chowdary Date: Thu, 5 Nov 2026 18:56:32 +0530 Subject: [PATCH] feat(source): compile an Amazon S3 prefix into the local wiki Bindings stay in config and credentials stay in env files, so a remote object can recompile through the existing add path without a new storage backend. Co-authored-by: Cursor --- .env.example | 13 + AGENTS.md | 3 + README.md | 16 ++ frontend/src/components/ConnectorCards.tsx | 259 +++++++++++++++-- frontend/src/locales/en/common.json | 20 +- frontend/src/locales/en/kb.json | 2 +- frontend/src/locales/en/settings.json | 2 +- frontend/src/locales/zh/common.json | 18 +- frontend/src/locales/zh/kb.json | 2 +- frontend/src/locales/zh/settings.json | 2 +- frontend/src/pages/KbDetail.tsx | 4 +- openkb/api.py | 2 + openkb/api_connectors_router.py | 175 ++++++++++++ openkb/cli.py | 42 ++- openkb/connectors/__init__.py | 4 + openkb/connectors/backend.py | 33 +++ openkb/connectors/credentials.py | 74 +++++ openkb/connectors/model.py | 67 +++++ openkb/connectors/s3.py | 109 ++++++++ openkb/connectors/state.py | 117 ++++++++ openkb/connectors/store.py | 159 +++++++++++ openkb/connectors/sync.py | 306 +++++++++++++++++++++ openkb/converter.py | 11 +- openkb/locks.py | 51 +++- openkb/source_cmd.py | 119 ++++++++ pyproject.toml | 3 + tests/test_api_connectors.py | 94 +++++++ tests/test_connectors.py | 299 ++++++++++++++++++++ tests/test_source_cli.py | 99 +++++++ uv.lock | 71 ++++- 30 files changed, 2112 insertions(+), 64 deletions(-) create mode 100644 openkb/api_connectors_router.py create mode 100644 openkb/connectors/__init__.py create mode 100644 openkb/connectors/backend.py create mode 100644 openkb/connectors/credentials.py create mode 100644 openkb/connectors/model.py create mode 100644 openkb/connectors/s3.py create mode 100644 openkb/connectors/state.py create mode 100644 openkb/connectors/store.py create mode 100644 openkb/connectors/sync.py create mode 100644 openkb/source_cmd.py create mode 100644 tests/test_api_connectors.py create mode 100644 tests/test_connectors.py create mode 100644 tests/test_source_cli.py diff --git a/.env.example b/.env.example index e0068c1f..7e99fa37 100644 --- a/.env.example +++ b/.env.example @@ -11,3 +11,16 @@ LLM_API_KEY=your-key-here # --- REST API server variables (read by `openkb-api` / `python -m openkb.api`) --- # OPENKB_API_TOKEN=... # required bearer token clients send as Authorization: Bearer # OPENKB_KB_ROOT=... # optional; where REST /init creates KBs (default: ~/.config/openkb/kbs) + +# --- Remote data sources (optional extras; not read from config.yaml) --- +# Amazon S3 — pip install "openkb[s3]" +OPENKB_S3_ACCESS_KEY_ID= +OPENKB_S3_SECRET_ACCESS_KEY= +# Google Drive — OAuth client you register. The refresh token is written by `openkb source add`, not by hand. +OPENKB_GDRIVE_CLIENT_ID= +OPENKB_GDRIVE_CLIENT_SECRET= +OPENKB_GDRIVE_REFRESH_TOKEN= +# OneDrive — same rule for the refresh token. +OPENKB_ONEDRIVE_CLIENT_ID= +OPENKB_ONEDRIVE_CLIENT_SECRET= +OPENKB_ONEDRIVE_REFRESH_TOKEN= diff --git a/AGENTS.md b/AGENTS.md index 0bda2d10..c922e2f8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,6 +20,9 @@ humans steer, agents execute. Optimize changes for agent legibility. - `config.py` — config loading/validation (LiteLLM passthrough, env). - `converter.py` — document → markdown conversion (markitdown). - `url_ingest.py` — fetch & ingest URLs (trafilatura). +- `source_cmd.py` — `openkb source` remote bindings (S3). +- `connectors/` — list, download, and sync a remote prefix through `add_single_file`. +- `api_connectors_router.py` — REST mirror of `openkb source`. - `images.py` — figure/image extraction & handling. - `indexer.py` — PageIndex tree indexing for long docs. - `mutation.py` — crash-safe, serial KB mutations. diff --git a/README.md b/README.md index 988bebda..70cd1e92 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,22 @@ Open `http://127.0.0.1:7566/` for the Workbench. Auth is off by default (local-f > Working on the UI itself? Run the Vite dev server with `cd frontend && npm install && npm run dev` (it proxies `/api` to a running `openkb-web`), or `npm run build` to regenerate the bundled `openkb/web/`. +### Amazon S3 + +Bind a bucket prefix and compile those objects into the local wiki. The SDK is an optional extra. Credentials use the CLI order: process environment, then `/.env`, then `~/.config/openkb/.env`. + +```bash +pip install "openkb[s3]" +# /.env +# OPENKB_S3_ACCESS_KEY_ID=... +# OPENKB_S3_SECRET_ACCESS_KEY=... + +openkb source add s3 --name papers --region us-east-1 --bucket my-bucket --prefix papers/ +openkb source sync papers +``` + +Google Drive and OneDrive are not available yet. An empty `--prefix` is the bucket root. `openkb source remove` deletes the binding and leaves compiled pages in place. + # 🧩 How OpenKB Works ### Architecture diff --git a/frontend/src/components/ConnectorCards.tsx b/frontend/src/components/ConnectorCards.tsx index 577a82eb..3c82f0ad 100644 --- a/frontend/src/components/ConnectorCards.tsx +++ b/frontend/src/components/ConnectorCards.tsx @@ -1,14 +1,14 @@ +import { useCallback, useEffect, useState } from "react" import { Cloud, HardDrive, ThumbsUp, type LucideIcon } from "lucide-react" import { useTranslation } from "react-i18next" +import { ApiError, apiFetch } from "@/api/client" /** - * Remote document-source connectors are NOT implemented (no OAuth/S3/sync - * backend). Rather than dead "coming soon" tiles, each card links to a GitHub - * issue search where users 👍 to signal demand — a telemetry-free, OSS-native - * way to prioritize. The copy MUST never imply a connector is connected. + * Remote document-source connectors. * - * `voteUrl` is an interim per-connector label search on the canonical repo; swap - * to a specific `.../issues/` once a tracking issue exists (one-line change). + * A provider with no backend keeps the GitHub vote link (`label:connector:`). + * Amazon S3 is shipped: with a knowledge base selected, zero bindings show a + * connect form and one or more bindings show Sync / Remove / Add another. */ export interface Connector { id: string @@ -21,38 +21,239 @@ const REPO = "https://github.com/VectifyAI/OpenKB" const voteSearch = (label: string) => `${REPO}/issues?q=${encodeURIComponent(`is:issue label:connector:${label}`)}` +const SHIPPED = new Set(["s3"]) + export const CONNECTORS: Connector[] = [ { id: "gdrive", label: "Google Drive", icon: Cloud, voteUrl: voteSearch("gdrive") }, { id: "s3", label: "Amazon S3", icon: HardDrive, voteUrl: voteSearch("s3") }, { id: "onedrive", label: "OneDrive", icon: Cloud, voteUrl: voteSearch("onedrive") }, ] -/** Grid of connector "want this?" cards linking to GitHub demand-vote searches. */ -export default function ConnectorCards() { +interface BindingView { + name: string + provider: string + description: string + stale: boolean + error: string | null +} + +const inputCls = + "mt-1 w-full h-8 rounded-md border border-input bg-transparent px-2 text-[12px] font-mono2 outline-none focus-visible:ring-2 focus-visible:ring-ring" + +function VoteCard({ connector }: { connector: Connector }) { const { t } = useTranslation("common") + const Icon = connector.icon return ( -
- {CONNECTORS.map((c) => ( - + + + +
+
{connector.label}
+
+ + {t("connector.vote")} +
+
+ + ) +} + +function S3Card({ kb }: { kb?: string }) { + const { t } = useTranslation("common") + const [bindings, setBindings] = useState([]) + const [adding, setAdding] = useState(false) + const [pending, setPending] = useState<"save" | "sync" | "remove" | null>(null) + const [error, setError] = useState(null) + const [name, setName] = useState("") + const [region, setRegion] = useState("") + const [bucket, setBucket] = useState("") + const [prefix, setPrefix] = useState("") + + const reload = useCallback(async () => { + if (!kb) return + const body = await apiFetch<{ connectors: BindingView[] }>( + `/api/v1/connectors?kb=${encodeURIComponent(kb)}`, + ) + setBindings(body.connectors.filter((item) => item.provider === "s3")) + }, [kb]) + + useEffect(() => { + if (!kb) return + let cancelled = false + reload().catch((exc: unknown) => { + if (!cancelled) setError(exc instanceof Error ? exc.message : String(exc)) + }) + return () => { + cancelled = true + } + }, [kb, reload]) + + const save = async () => { + if (!kb) return + setPending("save") + setError(null) + try { + await apiFetch("/api/v1/connectors", { + method: "POST", + body: { kb, name, provider: "s3", region, bucket, prefix }, + }) + setName("") + setRegion("") + setBucket("") + setPrefix("") + setAdding(false) + await reload() + } catch (exc) { + const message = exc instanceof ApiError ? exc.message : String(exc) + setError(t("connector.formError", { error: message })) + } finally { + setPending(null) + } + } + + const syncOne = async (bindingName: string) => { + if (!kb) return + setPending("sync") + setError(null) + try { + await apiFetch("/api/v1/connectors/sync", { + method: "POST", + body: { kb, name: bindingName, non_interactive: true }, + }) + await reload() + } catch (exc) { + setError(exc instanceof Error ? exc.message : String(exc)) + } finally { + setPending(null) + } + } + + const removeOne = async (bindingName: string) => { + if (!kb) return + setPending("remove") + setError(null) + try { + await apiFetch("/api/v1/connectors/remove", { + method: "POST", + body: { kb, name: bindingName }, + }) + await reload() + } catch (exc) { + setError(exc instanceof Error ? exc.message : String(exc)) + } finally { + setPending(null) + } + } + + const showForm = Boolean(kb) && (bindings.length === 0 || adding) + + return ( +
+
+ + + +
Amazon S3
+
+ {!kb &&

{t("connector.needKb")}

} + {kb && bindings.length > 0 && ( +
    + {bindings.map((binding) => ( +
  • +
    {binding.name}
    +
    {binding.description}
    +
    + {binding.stale ? t("connector.notSynced") : t("connector.synced")} +
    + {binding.error &&
    {binding.error}
    } +
    + + +
    +
    • + ))} +
+ )} + {kb && bindings.length > 0 && !adding && ( + + )} + {showForm && ( +
{ + event.preventDefault() + void save() + }} > - - - -
-
{c.label}
-
- - {t("connector.vote")} -
-
- - ))} + + + + + +
+ )} + {error &&

{error}

} +
+ ) +} + +/** Grid of connector cards. Pass `kb` on a knowledge-base page so S3 can bind. */ +export default function ConnectorCards({ kb }: { kb?: string }) { + return ( +
+ {CONNECTORS.map((connector) => + SHIPPED.has(connector.id) ? ( + + ) : ( + + ), + )}
) } diff --git a/frontend/src/locales/en/common.json b/frontend/src/locales/en/common.json index 91430075..dede5ceb 100644 --- a/frontend/src/locales/en/common.json +++ b/frontend/src/locales/en/common.json @@ -54,8 +54,24 @@ "baseLabel": "API address (optional; leave blank to use the current origin)", "connect": "Connect" }, - "connector": { + "connector": { "tooltip": "Cloud sync is in development · vote for it on GitHub", - "vote": "Want it? Vote" + "vote": "Want it? Vote", + "connect": "Connect", + "addAnother": "Add another", + "sync": "Sync", + "remove": "Remove", + "name": "Name", + "region": "Region", + "bucket": "Bucket", + "prefix": "Prefix", + "prefixHint": "Empty prefix is the whole bucket", + "notSynced": "Not synced since this change", + "synced": "Synced", + "save": "Save binding", + "saving": "Saving…", + "needKb": "Open a knowledge base to connect Amazon S3.", + "missingExtra": "Install the s3 extra on the server: pip install \"openkb[s3]\"", + "formError": "Could not save the binding: {{error}}" } } diff --git a/frontend/src/locales/en/kb.json b/frontend/src/locales/en/kb.json index d3543146..484e3cd0 100644 --- a/frontend/src/locales/en/kb.json +++ b/frontend/src/locales/en/kb.json @@ -82,7 +82,7 @@ }, "remote": { "heading": "Remote data sources", - "note": "Cloud connectors are in development and not yet available · want one? Click a card to vote on GitHub" + "note": "Amazon S3 connects on this page. Google Drive and OneDrive are not available yet — click a card to vote on GitHub." }, "overview": { "index": { "label": "Index", "caption": "wiki root" }, diff --git a/frontend/src/locales/en/settings.json b/frontend/src/locales/en/settings.json index 30ccd99b..5790f048 100644 --- a/frontend/src/locales/en/settings.json +++ b/frontend/src/locales/en/settings.json @@ -29,7 +29,7 @@ "kbRootEnvPinned": "Pinned by the OPENKB_KB_ROOT environment variable; edits here won't take effect until it's unset." }, "conn": { - "note": "Cloud data-source connectors (OAuth / S3) are in development and not yet available; for now, upload local files from the KB detail page. Want a connector? Click a card below to vote on GitHub and help us prioritize." + "note": "Amazon S3 is connected from a knowledge base page. Google Drive and OneDrive are not available yet — click a card to vote on GitHub." }, "about": { "links": { diff --git a/frontend/src/locales/zh/common.json b/frontend/src/locales/zh/common.json index 09edf636..67b9963e 100644 --- a/frontend/src/locales/zh/common.json +++ b/frontend/src/locales/zh/common.json @@ -56,6 +56,22 @@ }, "connector": { "tooltip": "云端同步开发中 · 去 GitHub 投票表达需求", - "vote": "想要它？去投票" + "vote": "想要它？去投票", + "connect": "连接", + "addAnother": "再添加一个", + "sync": "同步", + "remove": "移除", + "name": "名称", + "region": "区域", + "bucket": "存储桶", + "prefix": "前缀", + "prefixHint": "前缀留空表示整个存储桶", + "notSynced": "更改后尚未同步", + "synced": "已同步", + "save": "保存绑定", + "saving": "正在保存…", + "needKb": "请打开一个知识库以连接 Amazon S3。", + "missingExtra": "请在服务器安装 s3 扩展：pip install \"openkb[s3]\"", + "formError": "无法保存绑定：{{error}}" } } diff --git a/frontend/src/locales/zh/kb.json b/frontend/src/locales/zh/kb.json index f93280cd..6b02b5d9 100644 --- a/frontend/src/locales/zh/kb.json +++ b/frontend/src/locales/zh/kb.json @@ -82,7 +82,7 @@ }, "remote": { "heading": "远程数据源", - "note": "云端连接器开发中，尚不可用 · 想要它？点卡片去 GitHub 投票" + "note": "Amazon S3 可在本页连接。Google Drive 与 OneDrive 尚不可用 — 点卡片去 GitHub 投票。" }, "overview": { "index": { "label": "索引", "caption": "wiki 根" }, diff --git a/frontend/src/locales/zh/settings.json b/frontend/src/locales/zh/settings.json index 5d60a145..4cce991e 100644 --- a/frontend/src/locales/zh/settings.json +++ b/frontend/src/locales/zh/settings.json @@ -29,7 +29,7 @@ "kbRootEnvPinned": "已由环境变量 OPENKB_KB_ROOT 固定，此处的修改在取消该环境变量前不会生效。" }, "conn": { - "note": "云端数据源连接器（OAuth / S3）开发中，尚不可用；当前请在知识库详情页手动上传本地文件。想要某个连接器？点下方卡片去 GitHub 投票，帮我们排优先级。" + "note": "Amazon S3 在知识库页面连接。Google Drive 与 OneDrive 尚不可用 — 点下方卡片去 GitHub 投票。" }, "about": { "links": { diff --git a/frontend/src/pages/KbDetail.tsx b/frontend/src/pages/KbDetail.tsx index c4c5935e..00af3087 100644 --- a/frontend/src/pages/KbDetail.tsx +++ b/frontend/src/pages/KbDetail.tsx @@ -1266,13 +1266,13 @@ function DocumentsPane({ ))}
- {/* Remote connectors: no backend. Reframed as GitHub feature-request voting; never fake a connected state. */} + {/* S3 binds on this page. Drive and OneDrive stay vote links until they ship. */}

{t('kb:remote.heading')}

{t('kb:remote.note')}

- +
diff --git a/openkb/api.py b/openkb/api.py index 16b5db87..ab3cc2d7 100644 --- a/openkb/api.py +++ b/openkb/api.py @@ -31,6 +31,7 @@ from openkb.agent.query import build_run_config_from_bundle, run_query from openkb.api_config import apply_kb_config_patch, read_kb_config from openkb.api_config_router import config_router +from openkb.api_connectors_router import connectors_router from openkb.api_documents_router import documents_router from openkb.api_graph import graph_router from openkb.api_helpers import ( @@ -169,6 +170,7 @@ async def lifespan(app: FastAPI): app.include_router(kbs_router) app.include_router(pages_router) app.include_router(documents_router) + app.include_router(connectors_router) @app.get("/api/v1/kbs", response_model=KbListResponse) async def list_kbs_endpoint( diff --git a/openkb/api_connectors_router.py b/openkb/api_connectors_router.py new file mode 100644 index 00000000..6ccf83dd --- /dev/null +++ b/openkb/api_connectors_router.py @@ -0,0 +1,175 @@ +"""REST mirror of ``openkb source``. + +Lock-holding work runs in a worker thread. ``kb_lock`` reentrancy is +per-thread, so these handlers must not take the lock on the event loop. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field +from starlette.concurrency import run_in_threadpool + +from openkb.api_helpers import _resolve_kb, require_bearer_token +from openkb.connectors.credentials import s3_credentials +from openkb.connectors.model import parse_s3_binding +from openkb.connectors.state import KbBusyError +from openkb.connectors.store import remove_binding, upsert_binding +from openkb.connectors.sync import binding_views, sync_bindings + +connectors_router = APIRouter() + + +class ConnectorView(BaseModel): + name: str + provider: str + location: dict[str, str] + description: str + stale: bool + last_sync_at: str | None = None + added: int = 0 + skipped: int = 0 + updated: int = 0 + failed: int = 0 + error: str | None = None + + +class ConnectorListResponse(BaseModel): + connectors: list[ConnectorView] + + +class ConnectorAddRequest(BaseModel): + kb: str = Field(..., min_length=1) + name: str + provider: str = "s3" + region: str = "" + bucket: str = "" + prefix: str = "" + + +class ConnectorRemoveRequest(BaseModel): + kb: str = Field(..., min_length=1) + name: str + + +class ConnectorSyncRequest(BaseModel): + kb: str = Field(..., min_length=1) + name: str | None = None + non_interactive: bool = True + + +class BindingSyncView(BaseModel): + name: str + provider: str + ok: bool + added: int + skipped: int + updated: int + failed: int + error: str | None = None + + +class ConnectorSyncResponse(BaseModel): + exit_code: int + results: list[BindingSyncView] + + +class ConnectorMessage(BaseModel): + message: str + + +def _add(kb_dir, request: ConnectorAddRequest) -> str: + if request.provider != "s3": + raise ValueError("Only the s3 provider is available.") + if s3_credentials(kb_dir) is None: + raise ValueError( + "S3 credentials are missing. Set OPENKB_S3_ACCESS_KEY_ID and " + "OPENKB_S3_SECRET_ACCESS_KEY in the environment or /.env." + ) + binding = parse_s3_binding( + name=request.name, + region=request.region, + bucket=request.bucket, + prefix=request.prefix, + ) + return upsert_binding(kb_dir, binding) + + +@connectors_router.get("/api/v1/connectors", response_model=ConnectorListResponse) +async def list_connectors_endpoint( + kb: str = Query(...), + _: None = Depends(require_bearer_token), +) -> ConnectorListResponse: + kb_dir = _resolve_kb(kb) + views = await run_in_threadpool(binding_views, kb_dir) + return ConnectorListResponse(connectors=[ConnectorView(**view) for view in views]) + + +@connectors_router.post("/api/v1/connectors", response_model=ConnectorMessage) +async def add_connector_endpoint( + request: ConnectorAddRequest, + _: None = Depends(require_bearer_token), +) -> ConnectorMessage: + kb_dir = _resolve_kb(request.kb) + try: + message = await run_in_threadpool(_add, kb_dir, request) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except KbBusyError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return ConnectorMessage(message=message) + + +@connectors_router.post("/api/v1/connectors/remove", response_model=ConnectorMessage) +async def remove_connector_endpoint( + request: ConnectorRemoveRequest, + _: None = Depends(require_bearer_token), +) -> ConnectorMessage: + kb_dir = _resolve_kb(request.kb) + + def _remove(): + return remove_binding(kb_dir, request.name) + + try: + message = await run_in_threadpool(_remove) + except KeyError as exc: + raise HTTPException(status_code=404, detail=f"No binding named {request.name!r}.") from exc + except KbBusyError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return ConnectorMessage(message=message) + + +@connectors_router.post("/api/v1/connectors/sync", response_model=ConnectorSyncResponse) +async def sync_connectors_endpoint( + request: ConnectorSyncRequest, + _: None = Depends(require_bearer_token), +) -> ConnectorSyncResponse: + kb_dir = _resolve_kb(request.kb) + + def _sync(): + return sync_bindings( + kb_dir, + name=request.name, + non_interactive=request.non_interactive, + ) + + try: + outcome = await run_in_threadpool(_sync) + except KbBusyError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return ConnectorSyncResponse( + exit_code=outcome.exit_code, + results=[ + BindingSyncView( + name=item.name, + provider=item.provider, + ok=item.ok, + added=item.added, + skipped=item.skipped, + updated=item.updated, + failed=item.failed, + error=item.error, + ) + for item in outcome.reports + ], + ) diff --git a/openkb/cli.py b/openkb/cli.py index c9e54318..8b50ca47 100644 --- a/openkb/cli.py +++ b/openkb/cli.py @@ -445,15 +445,31 @@ def _run_compile_with_retry(coro_factory, label: str) -> None: def add_single_file( - file_path: Path, kb_dir: Path, *, stage: bool = True, bundle=None + file_path: Path, + kb_dir: Path, + *, + stage: bool = True, + bundle=None, + identity: str | None = None, ) -> Literal["added", "skipped", "failed"]: - """Convert, index, and compile a single document under the KB mutation lock.""" + """Convert, index, and compile a single document under the KB mutation lock. + + ``identity`` is a stable registry key for a remote object (for example + ``s3://bucket/key``). Local adds leave it unset and keep using the file path. + """ with kb_ingest_lock(kb_dir / ".openkb"): - return _add_single_file_locked(file_path, kb_dir, stage=stage, bundle=bundle) + return _add_single_file_locked( + file_path, kb_dir, stage=stage, bundle=bundle, identity=identity + ) def _add_single_file_locked( - file_path: Path, kb_dir: Path, *, stage: bool = True, bundle=None + file_path: Path, + kb_dir: Path, + *, + stage: bool = True, + bundle=None, + identity: str | None = None, ) -> Literal["added", "skipped", "failed"]: """Convert, index, and compile a single document into the knowledge base. @@ -487,7 +503,11 @@ def _add_single_file_locked( # 2. Convert document into staging when possible. click.echo(f"Adding: {file_path.name}") try: - result = convert_document(file_path, kb_dir, staging_dir=staging_dir) + # Omit the keyword when unset so local adds keep the historical call. + if identity is None: + result = convert_document(file_path, kb_dir, staging_dir=staging_dir) + else: + result = convert_document(file_path, kb_dir, staging_dir=staging_dir, identity=identity) except Exception as exc: click.echo(f" [ERROR] Conversion failed: {exc}") logger.debug("Conversion traceback:", exc_info=True) @@ -587,7 +607,7 @@ def commit_body(snapshot) -> None: "name": file_path.name, "doc_name": doc_name, "type": doc_type, - "path": _registry_path(file_path, kb_dir), + "path": identity if identity else _registry_path(file_path, kb_dir), } if result.raw_path is not None: meta["raw_path"] = _registry_path(result.raw_path, kb_dir) @@ -3806,3 +3826,13 @@ async def run_lint_report( "lint_files_changed": lint_files_changed, "lint_ghosts_removed": lint_ghosts_removed, } + + +def _register_source_command() -> None: + """Attach ``openkb source`` without importing it while this module loads.""" + from openkb.source_cmd import source as source_command + + cli.add_command(source_command) + + +_register_source_command() diff --git a/openkb/connectors/__init__.py b/openkb/connectors/__init__.py new file mode 100644 index 00000000..9f6c96a9 --- /dev/null +++ b/openkb/connectors/__init__.py @@ -0,0 +1,4 @@ +"""Remote data-source connectors. + +S3 is the first backend. Drive and OneDrive reuse this package later. +""" diff --git a/openkb/connectors/backend.py b/openkb/connectors/backend.py new file mode 100644 index 00000000..03298959 --- /dev/null +++ b/openkb/connectors/backend.py @@ -0,0 +1,33 @@ +"""List-and-download interface shared by remote backends.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + + +class TransientError(Exception): + """A retryable HTTP failure (throttle, timeout, 5xx).""" + + +class ProviderError(Exception): + """A failure of one provider. Other providers still run.""" + + +@dataclass(frozen=True) +class RemoteObject: + """One remote file selected for ingest.""" + + identity: str + name: str + + +class ObjectBackend(Protocol): + """List objects under a binding and download one of them.""" + + def list_objects(self) -> list[RemoteObject]: + """Return every object under the binding.""" + + def download(self, obj: RemoteObject, dest: Path) -> None: + """Write ``obj`` to ``dest``.""" diff --git a/openkb/connectors/credentials.py b/openkb/connectors/credentials.py new file mode 100644 index 00000000..a3098620 --- /dev/null +++ b/openkb/connectors/credentials.py @@ -0,0 +1,74 @@ +"""Connector credentials in CLI order. + +Process environment, then ``/.env``, then ``~/.config/openkb/.env``. +This is intentionally not :func:`openkb.config.resolve_credential_bundle`, +which lets the KB file win over the process environment. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from openkb.config import GLOBAL_CONFIG_DIR + +S3_ACCESS_KEY_ID = "OPENKB_S3_ACCESS_KEY_ID" +S3_SECRET_ACCESS_KEY = "OPENKB_S3_SECRET_ACCESS_KEY" +GDRIVE_CLIENT_ID = "OPENKB_GDRIVE_CLIENT_ID" +GDRIVE_CLIENT_SECRET = "OPENKB_GDRIVE_CLIENT_SECRET" +GDRIVE_REFRESH_TOKEN = "OPENKB_GDRIVE_REFRESH_TOKEN" +ONEDRIVE_CLIENT_ID = "OPENKB_ONEDRIVE_CLIENT_ID" +ONEDRIVE_CLIENT_SECRET = "OPENKB_ONEDRIVE_CLIENT_SECRET" +ONEDRIVE_REFRESH_TOKEN = "OPENKB_ONEDRIVE_REFRESH_TOKEN" + +CONNECTOR_ENV_KEYS = ( + S3_ACCESS_KEY_ID, + S3_SECRET_ACCESS_KEY, + GDRIVE_CLIENT_ID, + GDRIVE_CLIENT_SECRET, + GDRIVE_REFRESH_TOKEN, + ONEDRIVE_CLIENT_ID, + ONEDRIVE_CLIENT_SECRET, + ONEDRIVE_REFRESH_TOKEN, +) + + +def _dotenv_get(path: Path, key: str) -> str | None: + if not path.is_file(): + return None + from dotenv import dotenv_values + + value = dotenv_values(path).get(key) + if value is None or value == "": + return None + return value + + +def resolve_connector_credential(kb_dir: Path, key: str) -> str | None: + """Return the first non-empty value for ``key`` in CLI order.""" + process = os.environ.get(key) + if process: + return process + kb_value = _dotenv_get(kb_dir / ".env", key) + if kb_value: + return kb_value + return _dotenv_get(GLOBAL_CONFIG_DIR / ".env", key) + + +def s3_credentials(kb_dir: Path) -> tuple[str, str] | None: + """Return the S3 key pair, or ``None`` when either key is absent.""" + access = resolve_connector_credential(kb_dir, S3_ACCESS_KEY_ID) + secret = resolve_connector_credential(kb_dir, S3_SECRET_ACCESS_KEY) + if not access or not secret: + return None + return access, secret + + +def resolved_secrets(kb_dir: Path) -> list[str]: + """Credential values currently visible, used to keep them out of state files.""" + found: list[str] = [] + for key in CONNECTOR_ENV_KEYS: + value = resolve_connector_credential(kb_dir, key) + if value: + found.append(value) + return found diff --git a/openkb/connectors/model.py b/openkb/connectors/model.py new file mode 100644 index 00000000..37168ca4 --- /dev/null +++ b/openkb/connectors/model.py @@ -0,0 +1,67 @@ +"""Binding records and remote object identities.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}") + + +@dataclass(frozen=True) +class S3Binding: + """One Amazon S3 prefix bound to a knowledge base.""" + + name: str + region: str + bucket: str + prefix: str = "" + + provider: str = "s3" + + def location(self) -> dict[str, str]: + return {"region": self.region, "bucket": self.bucket, "prefix": self.prefix} + + def describe(self) -> str: + prefix = self.prefix or "" + return f"s3://{self.bucket}/{prefix}" + + +def s3_object_identity(bucket: str, key: str) -> str: + """Registry identity for one S3 object. A rename is a different key.""" + return f"s3://{bucket}/{key}" + + +def validate_binding_name(name: str) -> str: + cleaned = name.strip() + if not _NAME_RE.fullmatch(cleaned): + raise ValueError( + "Binding name must be 1-64 characters of letters, digits, '_' or '-', " + "and must start with a letter or digit." + ) + return cleaned + + +def parse_s3_binding( + *, + name: str, + region: str, + bucket: str, + prefix: str = "", +) -> S3Binding: + cleaned_name = validate_binding_name(name) + cleaned_region = region.strip() + cleaned_bucket = bucket.strip() + if not cleaned_region: + raise ValueError("S3 region is required.") + if not cleaned_bucket or "/" in cleaned_bucket: + raise ValueError("S3 bucket is required and cannot contain '/'.") + cleaned_prefix = prefix.strip() + if cleaned_prefix.startswith("/"): + raise ValueError("S3 prefix must not start with '/'. An empty prefix is the bucket root.") + return S3Binding( + name=cleaned_name, + region=cleaned_region, + bucket=cleaned_bucket, + prefix=cleaned_prefix, + ) diff --git a/openkb/connectors/s3.py b/openkb/connectors/s3.py new file mode 100644 index 00000000..a8e546c2 --- /dev/null +++ b/openkb/connectors/s3.py @@ -0,0 +1,109 @@ +"""Amazon S3 list and download. boto3 is imported only when a client is built.""" + +from __future__ import annotations + +from pathlib import Path + +from openkb.connectors.backend import ProviderError, RemoteObject, TransientError +from openkb.connectors.model import S3Binding, s3_object_identity + +_TRANSIENT_STATUS = {429, 500, 502, 503, 504} +_TRANSIENT_CODES = {"SlowDown", "RequestTimeout", "Throttling", "RequestTimeoutException"} + + +def _load_boto3(): + try: + import boto3 # type: ignore[import-not-found] + except ImportError as exc: + raise ProviderError( + 'Amazon S3 support is not installed. Install it with: pip install "openkb[s3]"' + ) from exc + return boto3 + + +def _raise_translated(exc: Exception) -> None: + try: + from botocore.exceptions import BotoCoreError, ClientError # type: ignore[import-not-found] + except ImportError: + raise ProviderError(str(exc)) from exc + if isinstance(exc, ClientError): + error = exc.response.get("Error", {}) if isinstance(exc.response, dict) else {} + meta = exc.response.get("ResponseMetadata", {}) if isinstance(exc.response, dict) else {} + status = meta.get("HTTPStatusCode") + code = str(error.get("Code", "")) + if status in _TRANSIENT_STATUS or code in _TRANSIENT_CODES: + raise TransientError(f"S3 transient error ({code or status})") from exc + raise ProviderError(f"S3 error ({code or status})") from exc + if isinstance(exc, BotoCoreError): + name = type(exc).__name__ + if "Timeout" in name or "Connection" in name: + raise TransientError(f"S3 transient error ({name})") from exc + raise ProviderError(f"S3 error ({name})") from exc + raise exc + + +class S3Backend: + """List and download one bucket prefix with caller-supplied keys.""" + + def __init__( + self, + binding: S3Binding, + access_key: str, + secret_key: str, + *, + client=None, + ) -> None: + self._binding = binding + self._access_key = access_key + self._secret_key = secret_key + self._client = client + + def _s3(self): + if self._client is not None: + return self._client + boto3 = _load_boto3() + return boto3.client( + "s3", + region_name=self._binding.region, + aws_access_key_id=self._access_key, + aws_secret_access_key=self._secret_key, + ) + + def list_objects(self) -> list[RemoteObject]: + client = self._s3() + kwargs: dict[str, str] = {"Bucket": self._binding.bucket} + if self._binding.prefix: + kwargs["Prefix"] = self._binding.prefix + found: list[RemoteObject] = [] + try: + paginator = client.get_paginator("list_objects_v2") + for page in paginator.paginate(**kwargs): + for item in page.get("Contents") or []: + key = item.get("Key") + if not isinstance(key, str) or not key or key.endswith("/"): + continue + name = key.rsplit("/", 1)[-1] + found.append( + RemoteObject( + identity=s3_object_identity(self._binding.bucket, key), + name=name, + ) + ) + except Exception as exc: + if isinstance(exc, (ProviderError, TransientError)): + raise + _raise_translated(exc) + return found + + def download(self, obj: RemoteObject, dest: Path) -> None: + prefix = f"s3://{self._binding.bucket}/" + if not obj.identity.startswith(prefix): + raise ProviderError(f"Object {obj.identity} is outside bucket {self._binding.bucket}.") + key = obj.identity[len(prefix) :] + dest.parent.mkdir(parents=True, exist_ok=True) + try: + self._s3().download_file(self._binding.bucket, key, str(dest)) + except Exception as exc: + if isinstance(exc, (ProviderError, TransientError)): + raise + _raise_translated(exc) diff --git a/openkb/connectors/state.py b/openkb/connectors/state.py new file mode 100644 index 00000000..69a39cd7 --- /dev/null +++ b/openkb/connectors/state.py @@ -0,0 +1,117 @@ +"""Derived sync status in ``.openkb/connector-state.json``. + +The file is not the wiki. Writes go through :func:`atomic_write_json` while +the exclusive ingest lock is held, and they do not open a mutation journal +of their own (taking the lock does run the existing journal recovery). +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import portalocker + +from openkb.locks import atomic_write_json, kb_lock + +LOCK_TIMEOUT_SECONDS = 30 +BUSY_MESSAGE = ( + "knowledge base is busy (another `openkb add` or sync is running); retry when it finishes" +) + +_COUNT_KEYS = ("added", "skipped", "updated", "failed") + + +class SecretInStateError(ValueError): + """Refused to persist a state document that contains a credential.""" + + +class KbBusyError(TimeoutError): + """The ingest lock was not free within the connector wait.""" + + +def state_path(kb_dir: Path) -> Path: + return kb_dir / ".openkb" / "connector-state.json" + + +def read_state(kb_dir: Path) -> dict[str, Any]: + path = state_path(kb_dir) + if not path.is_file(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + if not isinstance(data, dict): + return {} + return data + + +def is_stale(provider: str, location: dict[str, str], entry: dict[str, Any] | None) -> bool: + """True when the saved result describes a different binding than ``location``.""" + if not entry or not entry.get("last_sync_at"): + return True + return entry.get("provider") != provider or entry.get("location") != location + + +def _reject_secrets(document: dict[str, Any], secrets: list[str]) -> None: + payload = json.dumps(document) + for secret in secrets: + if secret and secret in payload: + raise SecretInStateError("refusing to write connector state that contains a credential") + + +def update_state( + kb_dir: Path, + name: str, + entry: dict[str, Any] | None, + *, + secrets: list[str] | None = None, + timeout: float | None = None, +) -> None: + """Insert, replace, or delete one binding's status under the ingest lock. + + ``entry is None`` deletes the key. ``timeout`` defaults to 30 seconds. + """ + + def mutate(data: dict[str, Any]) -> None: + if entry is None: + data.pop(name, None) + else: + data[name] = entry + + _mutate(kb_dir, mutate, secrets=secrets or [], timeout=timeout) + + +def _mutate( + kb_dir: Path, + mutate, + *, + secrets: list[str], + timeout: float | None, +) -> None: + wait = LOCK_TIMEOUT_SECONDS if timeout is None else timeout + try: + with kb_lock(kb_dir / ".openkb", exclusive=True, timeout=wait): + data = read_state(kb_dir) + mutate(data) + _reject_secrets(data, secrets) + atomic_write_json(state_path(kb_dir), data) + except TimeoutError as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + except portalocker.exceptions.LockException as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + + +def fresh_entry(provider: str, location: dict[str, str]) -> dict[str, Any]: + """A status row with zero counts, used when a sync produced no file results.""" + entry: dict[str, Any] = { + "provider": provider, + "location": location, + "last_sync_at": None, + "error": None, + } + for key in _COUNT_KEYS: + entry[key] = 0 + return entry diff --git a/openkb/connectors/store.py b/openkb/connectors/store.py new file mode 100644 index 00000000..01fcfa74 --- /dev/null +++ b/openkb/connectors/store.py @@ -0,0 +1,159 @@ +"""The ``connectors`` list in ``.openkb/config.yaml``. + +This is desired state. Sync status lives in :mod:`openkb.connectors.state`. +Config and state are updated in one ingest-lock hold so a replace cannot +leave the previous location's counts on the new binding. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import portalocker +import yaml + +from openkb.config import save_config +from openkb.connectors.model import S3Binding, parse_s3_binding +from openkb.connectors.state import ( + BUSY_MESSAGE, + LOCK_TIMEOUT_SECONDS, + KbBusyError, + read_state, + state_path, +) +from openkb.locks import atomic_write_json, kb_lock +from openkb.state import HashRegistry + + +def config_path(kb_dir: Path) -> Path: + return kb_dir / ".openkb" / "config.yaml" + + +def _read_raw(path: Path) -> dict[str, Any]: + if not path.is_file(): + return {} + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + if not isinstance(data, dict): + raise ValueError("config.yaml must be a mapping.") + return data + + +def _parse_item(item: object) -> S3Binding | None: + if not isinstance(item, dict): + return None + provider = item.get("provider") + if provider != "s3": + return None + try: + return parse_s3_binding( + name=str(item.get("name", "")), + region=str(item.get("region", "")), + bucket=str(item.get("bucket", "")), + prefix=str(item.get("prefix", "") or ""), + ) + except ValueError: + return None + + +def load_bindings(kb_dir: Path) -> list[S3Binding]: + raw = _read_raw(config_path(kb_dir)) + items = raw.get("connectors") or [] + if not isinstance(items, list): + return [] + bindings: list[S3Binding] = [] + for item in items: + parsed = _parse_item(item) + if parsed is not None: + bindings.append(parsed) + return bindings + + +def _count_remote_pages(kb_dir: Path, bucket: str, prefix: str) -> int: + registry = HashRegistry(kb_dir / ".openkb" / "hashes.json") + needle = f"s3://{bucket}/{prefix}" + count = 0 + for meta in registry.all_entries().values(): + path = meta.get("path") + if isinstance(path, str) and path.startswith(needle): + count += 1 + return count + + +def upsert_binding(kb_dir: Path, binding: S3Binding) -> str: + """Insert or replace ``binding``. Returns ``created`` or ``replaced``. + + A replaced binding's sync status is cleared. Compiled pages stay. + """ + message = "created" + try: + with kb_lock(kb_dir / ".openkb", exclusive=True, timeout=LOCK_TIMEOUT_SECONDS): + path = config_path(kb_dir) + raw = _read_raw(path) + items = raw.get("connectors") + if not isinstance(items, list): + items = [] + kept: list[object] = [] + previous: S3Binding | None = None + for item in items: + parsed = _parse_item(item) + if parsed is not None and parsed.name == binding.name: + previous = parsed + continue + if isinstance(item, dict) and item.get("name") == binding.name: + previous = parsed + continue + kept.append(item) + kept.append( + { + "name": binding.name, + "provider": "s3", + "region": binding.region, + "bucket": binding.bucket, + "prefix": binding.prefix, + } + ) + raw["connectors"] = kept + save_config(path, raw) + if previous is not None: + data = read_state(kb_dir) + data.pop(binding.name, None) + atomic_write_json(state_path(kb_dir), data) + count = _count_remote_pages(kb_dir, previous.bucket, previous.prefix) + message = ( + f"Binding `{binding.name}` now points to {binding.describe()}. " + f"{count} documents compiled from the previous location remain in the wiki." + ) + else: + message = f"Binding `{binding.name}` saved ({binding.describe()})." + except TimeoutError as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + except portalocker.exceptions.LockException as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + return message + + +def remove_binding(kb_dir: Path, name: str) -> str: + """Delete the binding record and its status. Wiki pages and ``raw/`` stay.""" + try: + with kb_lock(kb_dir / ".openkb", exclusive=True, timeout=LOCK_TIMEOUT_SECONDS): + path = config_path(kb_dir) + raw = _read_raw(path) + items = raw.get("connectors") + if not isinstance(items, list): + items = [] + kept = [ + item for item in items if not (isinstance(item, dict) and item.get("name") == name) + ] + if len(kept) == len(items): + raise KeyError(name) + raw["connectors"] = kept + save_config(path, raw) + data = read_state(kb_dir) + data.pop(name, None) + atomic_write_json(state_path(kb_dir), data) + except TimeoutError as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + except portalocker.exceptions.LockException as exc: + raise KbBusyError(BUSY_MESSAGE) from exc + return f"Removed binding `{name}`. Documents compiled from it remain in the wiki." diff --git a/openkb/connectors/sync.py b/openkb/connectors/sync.py new file mode 100644 index 00000000..17e2ce3f --- /dev/null +++ b/openkb/connectors/sync.py @@ -0,0 +1,306 @@ +"""Sync saved bindings into the knowledge base through ``add_single_file``.""" + +from __future__ import annotations + +import tempfile +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path + +from openkb.connectors.backend import ObjectBackend, ProviderError, RemoteObject, TransientError +from openkb.connectors.credentials import resolved_secrets, s3_credentials +from openkb.connectors.model import S3Binding +from openkb.connectors.s3 import S3Backend +from openkb.connectors.state import ( + SecretInStateError, + fresh_entry, + is_stale, + read_state, + update_state, +) +from openkb.connectors.store import load_bindings +from openkb.state import HashRegistry + +RETRY_ATTEMPTS = 3 + +AddFn = Callable[..., str] +SignInFn = Callable[[], None] + + +@dataclass +class BindingReport: + name: str + provider: str + ok: bool + added: int = 0 + skipped: int = 0 + updated: int = 0 + failed: int = 0 + error: str | None = None + stale: bool = False + + +@dataclass +class SyncOutcome: + reports: list[BindingReport] = field(default_factory=list) + + @property + def exit_code(self) -> int: + for report in self.reports: + if not report.ok or report.failed: + return 1 + return 0 + + +def recover_auth(*, non_interactive: bool, sign_in: SignInFn | None) -> None: + """Run loopback sign-in, or fail the provider when that is not allowed. + + S3 does not call this. Drive and OneDrive will. ``--non-interactive`` never + invokes ``sign_in``. + """ + if non_interactive or sign_in is None: + raise ProviderError( + "Sign-in is required. Re-run without --non-interactive to open a browser." + ) + sign_in() + + +def _extensions() -> set[str]: + from openkb.cli import SUPPORTED_EXTENSIONS + + return SUPPORTED_EXTENSIONS + + +def _add_file(path: Path, kb_dir: Path, identity: str) -> str: + from openkb.cli import add_single_file + from openkb.locks import kb_ingest_lock_held + + if kb_ingest_lock_held(kb_dir / ".openkb"): + raise RuntimeError("sync called add_single_file while holding the ingest lock") + return add_single_file(path, kb_dir, identity=identity) + + +def _retry(action: Callable[[], object]) -> object: + last: TransientError | None = None + for _attempt in range(RETRY_ATTEMPTS): + try: + return action() + except TransientError as exc: + last = exc + assert last is not None + raise last + + +def _record( + kb_dir: Path, + binding: S3Binding, + *, + added: int, + skipped: int, + updated: int, + failed: int, + error: str | None, + synced: bool, +) -> None: + entry = fresh_entry(binding.provider, binding.location()) + entry["added"] = added + entry["skipped"] = skipped + entry["updated"] = updated + entry["failed"] = failed + entry["error"] = error + if synced: + entry["last_sync_at"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + secrets = resolved_secrets(kb_dir) + try: + update_state(kb_dir, binding.name, entry, secrets=secrets) + except SecretInStateError: + entry["error"] = "sync failed" + update_state(kb_dir, binding.name, entry, secrets=secrets) + + +def _sync_one( + kb_dir: Path, + binding: S3Binding, + backend: ObjectBackend, + add: AddFn, +) -> BindingReport: + added = skipped = updated = failed = 0 + error: str | None = None + try: + objects = _retry(backend.list_objects) + except (TransientError, ProviderError) as exc: + _record( + kb_dir, + binding, + added=0, + skipped=0, + updated=0, + failed=0, + error=str(exc), + synced=False, + ) + return BindingReport( + name=binding.name, + provider=binding.provider, + ok=False, + error=str(exc), + ) + assert isinstance(objects, list) + registry = HashRegistry(kb_dir / ".openkb" / "hashes.json") + allowed = _extensions() + for obj in objects: + if not isinstance(obj, RemoteObject): + continue + suffix = Path(obj.name).suffix.lower() + if suffix not in allowed: + skipped += 1 + continue + prior = registry.get_by_path(obj.identity) + try: + with tempfile.TemporaryDirectory(prefix="openkb-src-") as tmp: + dest = Path(tmp) / (Path(obj.name).name or "object") + + def download(obj: RemoteObject = obj, dest: Path = dest) -> None: + backend.download(obj, dest) + + _retry(download) + status = add(dest, kb_dir, identity=obj.identity) + except TransientError as exc: + failed += 1 + error = str(exc) + continue + except Exception as exc: + failed += 1 + error = str(exc) + continue + if status == "skipped": + skipped += 1 + elif status == "failed": + failed += 1 + error = f"Failed to compile {obj.name}." + elif prior is not None: + updated += 1 + else: + added += 1 + ok = failed == 0 + _record( + kb_dir, + binding, + added=added, + skipped=skipped, + updated=updated, + failed=failed, + error=error, + synced=True, + ) + return BindingReport( + name=binding.name, + provider=binding.provider, + ok=ok, + added=added, + skipped=skipped, + updated=updated, + failed=failed, + error=error, + ) + + +def sync_bindings( + kb_dir: Path, + *, + name: str | None = None, + non_interactive: bool = False, + sign_in: SignInFn | None = None, + add: AddFn | None = None, + backend_for: Callable[[S3Binding, str, str], ObjectBackend] | None = None, +) -> SyncOutcome: + """Sync every S3 binding, or the one named ``name``. + + ``sign_in`` is accepted so ``--non-interactive`` can be tested. S3 never + calls it. ``add`` and ``backend_for`` exist so tests can avoid the network + and the compiler. + """ + # S3 has no browser sign-in. The flag stays on the command so Drive can + # reuse it. A rejected token must not open a browser in non-interactive mode. + if non_interactive: + sign_in = None + del sign_in + bindings = load_bindings(kb_dir) + if name is not None: + bindings = [item for item in bindings if item.name == name] + if not bindings: + return SyncOutcome( + reports=[ + BindingReport( + name=name, + provider="s3", + ok=False, + error=f"No binding named {name!r}.", + ) + ] + ) + creds = s3_credentials(kb_dir) + reports: list[BindingReport] = [] + if bindings and creds is None: + for binding in bindings: + message = ( + "S3 credentials are missing. Set OPENKB_S3_ACCESS_KEY_ID and " + "OPENKB_S3_SECRET_ACCESS_KEY in the environment or /.env." + ) + _record( + kb_dir, + binding, + added=0, + skipped=0, + updated=0, + failed=0, + error=message, + synced=False, + ) + reports.append( + BindingReport( + name=binding.name, + provider="s3", + ok=False, + error=message, + ) + ) + return SyncOutcome(reports=reports) + assert creds is not None or not bindings + add_fn = add or _add_file + for binding in bindings: + assert creds is not None + access, secret = creds + backend = ( + backend_for(binding, access, secret) + if backend_for is not None + else S3Backend(binding, access, secret) + ) + reports.append(_sync_one(kb_dir, binding, backend, add_fn)) + return SyncOutcome(reports=reports) + + +def binding_views(kb_dir: Path) -> list[dict[str, object]]: + """Bindings plus the last sync row, for the CLI and the Workbench.""" + state = read_state(kb_dir) + views: list[dict[str, object]] = [] + for binding in load_bindings(kb_dir): + entry = state.get(binding.name) + row = entry if isinstance(entry, dict) else None + views.append( + { + "name": binding.name, + "provider": binding.provider, + "location": binding.location(), + "description": binding.describe(), + "stale": is_stale(binding.provider, binding.location(), row), + "last_sync_at": None if row is None else row.get("last_sync_at"), + "added": 0 if row is None else int(row.get("added") or 0), + "skipped": 0 if row is None else int(row.get("skipped") or 0), + "updated": 0 if row is None else int(row.get("updated") or 0), + "failed": 0 if row is None else int(row.get("failed") or 0), + "error": None if row is None else row.get("error"), + } + ) + return views diff --git a/openkb/converter.py b/openkb/converter.py index 08929e01..4e0b9eb0 100644 --- a/openkb/converter.py +++ b/openkb/converter.py @@ -77,6 +77,7 @@ def resolve_doc_name( registry: HashRegistry, *, persist_legacy: bool = True, + path_key: str | None = None, ) -> str: """Resolve the stable wiki name for ``src`` (Scheme A). @@ -87,7 +88,13 @@ def resolve_doc_name( sanitized stem unless another document already owns that name, in which case it gets a deterministic ``-{sha256(path)[:8]}`` suffix. """ - path_key = _registry_path(src, kb_dir) + explicit_identity = path_key is not None + if path_key is None: + path_key = _registry_path(src, kb_dir) + if explicit_identity: + # Remote objects have no filesystem path. Skip the legacy stem backfill + # so a cloud file cannot attach itself to an unrelated local document. + return resolve_doc_name_from_key(src.stem, path_key, registry) known = registry.get_by_path(path_key) if known is not None: @@ -143,6 +150,7 @@ def convert_document( kb_dir: Path, *, staging_dir: Path | None = None, + identity: str | None = None, ) -> ConvertResult: """Convert a document and integrate it into the knowledge base. @@ -181,6 +189,7 @@ def convert_document( kb_dir, registry, persist_legacy=staging_dir is None, + path_key=identity, ) # ------------------------------------------------------------------ diff --git a/openkb/locks.py b/openkb/locks.py index a085da75..37f40e53 100644 --- a/openkb/locks.py +++ b/openkb/locks.py @@ -14,13 +14,14 @@ import os import tempfile import threading +import time from pathlib import Path from typing import IO, Iterator import portalocker -def flock(fh: IO, *, exclusive: bool) -> None: +def flock(fh: IO, *, exclusive: bool, timeout: float | None = None) -> None: """Acquire an advisory lock on an open file handle (cross-platform). Delegates to :mod:`portalocker`: @@ -31,10 +32,29 @@ def flock(fh: IO, *, exclusive: bool) -> None: readers are honoured; exclusive locks use ``msvcrt.locking``, which retries for ~10s and then raises rather than blocking indefinitely. - On failure portalocker raises :class:`portalocker.LockException` — note this - is *not* an ``OSError`` (e.g. on filesystems without working lock support). + ``timeout`` bounds the wait. ``None`` keeps the historical blocking + acquire used by ``add``. On failure portalocker raises + :class:`portalocker.LockException` — note this is *not* an ``OSError`` + (e.g. on filesystems without working lock support, or a timed-out wait). """ - portalocker.lock(fh, portalocker.LOCK_EX if exclusive else portalocker.LOCK_SH) + flags = portalocker.LOCK_EX if exclusive else portalocker.LOCK_SH + if timeout is None: + portalocker.lock(fh, flags) + return + # portalocker 3.2's lock() has no timeout argument. Poll the non-blocking + # flag until the deadline, then raise the same LockException a contended + # non-blocking acquire raises. + deadline = time.monotonic() + timeout + nonblocking = flags | portalocker.LOCK_NB + while True: + try: + portalocker.lock(fh, nonblocking) + return + except portalocker.LockException: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise + time.sleep(min(0.05, remaining)) def funlock(fh: IO) -> None: @@ -68,10 +88,16 @@ def read(self) -> Iterator[None]: self._condition.notify_all() @contextlib.contextmanager - def write(self) -> Iterator[None]: + def write(self, timeout: float | None = None) -> Iterator[None]: + deadline = None if timeout is None else time.monotonic() + timeout with self._condition: while self._writer or self._readers: - self._condition.wait() + if deadline is None: + self._condition.wait() + continue + remaining = deadline - time.monotonic() + if remaining <= 0 or not self._condition.wait(timeout=remaining): + raise TimeoutError("timed out waiting for the knowledge base lock") self._writer = True try: yield @@ -126,8 +152,13 @@ def _drain_pending_journals(openkb_dir: Path) -> None: @contextlib.contextmanager -def kb_lock(openkb_dir: Path, *, exclusive: bool) -> Iterator[None]: - """Hold a KB-level advisory lock.""" +def kb_lock(openkb_dir: Path, *, exclusive: bool, timeout: float | None = None) -> Iterator[None]: + """Hold a KB-level advisory lock. + + ``timeout`` applies only to a fresh exclusive acquire. ``None`` blocks, which + is what ``add`` uses. A thread that already holds the lock reenters without + waiting. + """ lock_path = openkb_dir / "ingest.lock" lock_path.parent.mkdir(parents=True, exist_ok=True) resolved = lock_path.resolve() @@ -156,10 +187,10 @@ def kb_lock(openkb_dir: Path, *, exclusive: bool) -> Iterator[None]: return local_lock = _local_lock(lock_path) - local_context = local_lock.write() if exclusive else local_lock.read() + local_context = local_lock.write(timeout=timeout) if exclusive else local_lock.read() with local_context: with lock_path.open("a+", encoding="utf-8") as fh: - flock(fh, exclusive=exclusive) + flock(fh, exclusive=exclusive, timeout=timeout) held[resolved] = (1, 0) if exclusive else (0, 1) try: if exclusive: diff --git a/openkb/source_cmd.py b/openkb/source_cmd.py new file mode 100644 index 00000000..c6943914 --- /dev/null +++ b/openkb/source_cmd.py @@ -0,0 +1,119 @@ +"""``openkb source`` — bind a remote prefix and sync it into the knowledge base.""" + +from __future__ import annotations + +from pathlib import Path + +import click + +from openkb.connectors.credentials import s3_credentials +from openkb.connectors.model import parse_s3_binding +from openkb.connectors.state import KbBusyError +from openkb.connectors.store import remove_binding, upsert_binding +from openkb.connectors.sync import binding_views, sync_bindings + + +def _kb_dir(ctx: click.Context) -> Path: + from openkb.cli import _find_kb_dir + + root = ctx.find_root() + override = None + if isinstance(root.obj, dict): + override = root.obj.get("kb_dir_override") + found = _find_kb_dir(override) + if found is None: + raise click.ClickException("No knowledge base found. Run `openkb init` or pass --kb-dir.") + return found + + +@click.group() +def source() -> None: + """Bind a remote data source and sync it into this knowledge base.""" + + +@source.command("add") +@click.argument("provider", type=click.Choice(["s3"])) +@click.option("--name", required=True, help="Local name for this binding.") +@click.option("--region", default="", help="AWS region, for example us-east-1.") +@click.option("--bucket", default="", help="S3 bucket name.") +@click.option("--prefix", default="", help="Key prefix. Empty means the bucket root.") +@click.option( + "--non-interactive", + is_flag=True, + default=False, + help="Never open a browser. S3 does not open one either way.", +) +@click.pass_context +def add_cmd(ctx, provider, name, region, bucket, prefix, non_interactive) -> None: + """Save an S3 binding. Credentials must already be available.""" + del provider, non_interactive + kb_dir = _kb_dir(ctx) + if s3_credentials(kb_dir) is None: + raise click.ClickException( + "S3 credentials are missing. Set OPENKB_S3_ACCESS_KEY_ID and " + "OPENKB_S3_SECRET_ACCESS_KEY in the environment or /.env." + ) + try: + binding = parse_s3_binding(name=name, region=region, bucket=bucket, prefix=prefix) + message = upsert_binding(kb_dir, binding) + except (ValueError, KbBusyError) as exc: + raise click.ClickException(str(exc)) from exc + click.echo(message) + + +@source.command("list") +@click.pass_context +def list_cmd(ctx) -> None: + """List bindings and the last sync status.""" + kb_dir = _kb_dir(ctx) + views = binding_views(kb_dir) + if not views: + click.echo("No remote bindings.") + return + for view in views: + status = "not synced since this change" if view["stale"] else "synced" + click.echo(f"{view['name']}\t{view['provider']}\t{view['description']}\t{status}") + + +@source.command("remove") +@click.argument("name") +@click.pass_context +def remove_cmd(ctx, name) -> None: + """Remove a binding. Compiled documents stay in the wiki.""" + kb_dir = _kb_dir(ctx) + try: + message = remove_binding(kb_dir, name) + except KeyError as exc: + raise click.ClickException(f"No binding named {name!r}.") from exc + except KbBusyError as exc: + raise click.ClickException(str(exc)) from exc + click.echo(message) + + +@source.command("sync") +@click.argument("name", required=False) +@click.option( + "--non-interactive", + is_flag=True, + default=False, + help="Never open a browser. A missing credential fails that provider.", +) +@click.pass_context +def sync_cmd(ctx, name, non_interactive) -> None: + """Download new and changed objects and compile them.""" + kb_dir = _kb_dir(ctx) + try: + outcome = sync_bindings(kb_dir, name=name, non_interactive=non_interactive) + except KbBusyError as exc: + raise click.ClickException(str(exc)) from exc + for report in outcome.reports: + if report.error and not report.ok and report.added == 0 and report.failed == 0: + click.echo(f"{report.name}: {report.error}", err=True) + continue + click.echo( + f"{report.name}: added {report.added}, skipped {report.skipped}, " + f"updated {report.updated}, failed {report.failed}" + ) + if report.error: + click.echo(f" {report.error}", err=True) + ctx.exit(outcome.exit_code) diff --git a/pyproject.toml b/pyproject.toml index 389a8b2b..f4988c00 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,9 @@ dev = [ # `pip install "openkb[api]"` keeps working. web = ["fastapi", "uvicorn", "python-multipart"] api = ["openkb[web]"] +# Cloud SDKs stay out of the core install. Tests mock the client and do not +# import boto3, so CI's `uv sync --locked --extra dev --extra api` is unchanged. +s3 = ["boto3==1.43.124"] [tool.hatch.version] source = "vcs" diff --git a/tests/test_api_connectors.py b/tests/test_api_connectors.py new file mode 100644 index 00000000..80193be2 --- /dev/null +++ b/tests/test_api_connectors.py @@ -0,0 +1,94 @@ +"""REST connectors call the same functions as the CLI, off the event loop.""" + +from __future__ import annotations + +from fastapi.testclient import TestClient + +from openkb.api import create_app +from openkb.connectors.sync import BindingReport, SyncOutcome + + +def _client(monkeypatch, token: str | None = "secret") -> TestClient: + if token is None: + monkeypatch.delenv("OPENKB_API_TOKEN", raising=False) + else: + monkeypatch.setenv("OPENKB_API_TOKEN", token) + return TestClient(create_app()) + + +def _auth(token: str = "secret") -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def _kb(tmp_path): + kb = tmp_path / "kb" + (kb / ".openkb").mkdir(parents=True) + (kb / "wiki").mkdir() + (kb / ".env").write_text( + "OPENKB_S3_ACCESS_KEY_ID=AKIA_TEST\nOPENKB_S3_SECRET_ACCESS_KEY=secret-test\n", + encoding="utf-8", + ) + return kb + + +def _use_kb(monkeypatch, kb_dir, name: str = "test-kb") -> str: + def resolve(kb): + assert kb == name + return kb_dir + + monkeypatch.setattr("openkb.api_helpers.resolve_kb_alias", resolve) + return name + + +def test_connector_round_trip(tmp_path, monkeypatch): + kb = _kb(tmp_path) + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + name = _use_kb(monkeypatch, kb) + client = _client(monkeypatch) + created = client.post( + "/api/v1/connectors", + headers=_auth(), + json={ + "kb": name, + "name": "papers", + "provider": "s3", + "region": "us-east-1", + "bucket": "acme", + "prefix": "docs/", + }, + ) + assert created.status_code == 200, created.text + listed = client.get("/api/v1/connectors", headers=_auth(), params={"kb": name}) + assert listed.status_code == 200 + body = listed.json()["connectors"] + assert body[0]["name"] == "papers" + assert body[0]["location"]["bucket"] == "acme" + assert body[0]["stale"] is True + + def fake_sync(kb_dir, *, name=None, non_interactive=False, **kwargs): + assert non_interactive is True + return SyncOutcome( + reports=[ + BindingReport(name="papers", provider="s3", ok=True, added=1), + ] + ) + + monkeypatch.setattr("openkb.api_connectors_router.sync_bindings", fake_sync) + synced = client.post( + "/api/v1/connectors/sync", + headers=_auth(), + json={"kb": name, "name": "papers"}, + ) + assert synced.status_code == 200 + assert synced.json()["exit_code"] == 0 + removed = client.post( + "/api/v1/connectors/remove", + headers=_auth(), + json={"kb": name, "name": "papers"}, + ) + assert removed.status_code == 200 + assert "remain" in removed.json()["message"] + page = kb / "wiki" / "keep.md" + page.write_text("stay", encoding="utf-8") + assert page.read_text(encoding="utf-8") == "stay" diff --git a/tests/test_connectors.py b/tests/test_connectors.py new file mode 100644 index 00000000..c3fb1298 --- /dev/null +++ b/tests/test_connectors.py @@ -0,0 +1,299 @@ +"""Connector store, credentials, sync, and the S3 backend. No live AWS calls.""" + +from __future__ import annotations + +import threading +from pathlib import Path + +import pytest + +from openkb.connectors.backend import ProviderError, RemoteObject, TransientError +from openkb.connectors.credentials import resolve_connector_credential +from openkb.connectors.model import S3Binding, parse_s3_binding, s3_object_identity +from openkb.connectors.s3 import S3Backend, _load_boto3 +from openkb.connectors.state import ( + LOCK_TIMEOUT_SECONDS, + KbBusyError, + SecretInStateError, + read_state, + update_state, +) +from openkb.connectors.store import load_bindings, remove_binding, upsert_binding +from openkb.connectors.sync import recover_auth, sync_bindings +from openkb.locks import kb_ingest_lock, kb_ingest_lock_held +from openkb.state import HashRegistry + + +def _kb(tmp_path: Path) -> Path: + kb = tmp_path / "kb" + (kb / ".openkb").mkdir(parents=True) + (kb / "wiki").mkdir() + (kb / "raw").mkdir() + (kb / "wiki" / "summaries").mkdir() + return kb + + +def _binding(**kwargs) -> S3Binding: + values = {"name": "papers", "region": "us-east-1", "bucket": "acme", "prefix": "docs/"} + values.update(kwargs) + return parse_s3_binding(**values) + + +def _creds(kb: Path) -> None: + (kb / ".env").write_text( + "OPENKB_S3_ACCESS_KEY_ID=AKIA_TEST\nOPENKB_S3_SECRET_ACCESS_KEY=secret-test\n", + encoding="utf-8", + ) + + +class _Backend: + def __init__(self, objects: list[RemoteObject], *, download_failures: int = 0) -> None: + self.objects = objects + self.download_failures = download_failures + self.downloads = 0 + + def list_objects(self) -> list[RemoteObject]: + return list(self.objects) + + def download(self, obj: RemoteObject, dest: Path) -> None: + self.downloads += 1 + if self.downloads <= self.download_failures: + raise TransientError("slow down") + dest.write_bytes(b"bytes:" + obj.identity.encode()) + + +def test_credential_order_is_process_then_kb_then_global(tmp_path, monkeypatch): + kb = _kb(tmp_path) + global_dir = tmp_path / "global" + global_dir.mkdir() + monkeypatch.setattr("openkb.connectors.credentials.GLOBAL_CONFIG_DIR", global_dir) + (global_dir / ".env").write_text("OPENKB_S3_ACCESS_KEY_ID=from-global\n", encoding="utf-8") + (kb / ".env").write_text("OPENKB_S3_ACCESS_KEY_ID=from-kb\n", encoding="utf-8") + monkeypatch.setenv("OPENKB_S3_ACCESS_KEY_ID", "from-env") + + assert resolve_connector_credential(kb, "OPENKB_S3_ACCESS_KEY_ID") == "from-env" + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID") + assert resolve_connector_credential(kb, "OPENKB_S3_ACCESS_KEY_ID") == "from-kb" + (kb / ".env").write_text("", encoding="utf-8") + assert resolve_connector_credential(kb, "OPENKB_S3_ACCESS_KEY_ID") == "from-global" + + +def test_replace_clears_state_and_remove_keeps_wiki(tmp_path): + kb = _kb(tmp_path) + message = upsert_binding(kb, _binding()) + assert message.startswith("Binding `papers` saved") + update_state( + kb, + "papers", + { + "provider": "s3", + "location": _binding().location(), + "last_sync_at": "2026-11-05T00:00:00Z", + "added": 2, + "skipped": 0, + "updated": 0, + "failed": 0, + "error": None, + }, + ) + page = kb / "wiki" / "summaries" / "kept.md" + raw = kb / "raw" / "kept.pdf" + page.write_text("stay", encoding="utf-8") + raw.write_bytes(b"stay") + + replaced = upsert_binding(kb, _binding(bucket="other")) + assert "remain in the wiki" in replaced + assert "papers" not in read_state(kb) + + removed = remove_binding(kb, "papers") + assert "remain in the wiki" in removed + assert load_bindings(kb) == [] + assert page.read_text(encoding="utf-8") == "stay" + assert raw.read_bytes() == b"stay" + + +def test_state_refuses_a_credential(tmp_path): + kb = _kb(tmp_path) + with pytest.raises(SecretInStateError): + update_state( + kb, + "papers", + {"error": "leaked secret-test"}, + secrets=["secret-test"], + ) + assert read_state(kb) == {} + + +def test_state_write_times_out_while_add_holds_the_lock(tmp_path): + assert LOCK_TIMEOUT_SECONDS == 30 + kb = _kb(tmp_path) + started = threading.Event() + release = threading.Event() + + def holder() -> None: + with kb_ingest_lock(kb / ".openkb"): + started.set() + assert release.wait(timeout=2) + + thread = threading.Thread(target=holder) + thread.start() + assert started.wait(timeout=2) + try: + with pytest.raises(KbBusyError, match="knowledge base is busy"): + update_state( + kb, + "papers", + {"provider": "s3", "location": {}, "error": None}, + timeout=0.2, + ) + finally: + release.set() + thread.join(timeout=2) + + +def test_sync_skips_known_hash_and_updates_same_key(tmp_path, monkeypatch): + kb = _kb(tmp_path) + _creds(kb) + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + upsert_binding(kb, _binding(prefix="")) + identity = s3_object_identity("acme", "docs/a.txt") + HashRegistry(kb / ".openkb" / "hashes.json").add( + "old", + {"doc_name": "a", "path": identity, "name": "a.txt"}, + ) + calls: list[str] = [] + + def add(path: Path, kb_dir: Path, identity: str) -> str: + assert not kb_ingest_lock_held(kb_dir / ".openkb") + calls.append(identity) + if path.read_bytes() == b"same": + return "skipped" + return "added" + + same = RemoteObject(identity=identity, name="a.txt") + backend = _Backend([same]) + backend.download = lambda obj, dest: dest.write_bytes(b"same") # type: ignore[method-assign] + outcome = sync_bindings( + kb, + add=add, + backend_for=lambda binding, access, secret: backend, + ) + assert outcome.exit_code == 0 + assert outcome.reports[0].skipped == 1 + assert calls == [identity] + + backend.download = lambda obj, dest: dest.write_bytes(b"new") # type: ignore[method-assign] + outcome = sync_bindings( + kb, + add=add, + backend_for=lambda binding, access, secret: backend, + ) + assert outcome.reports[0].updated == 1 + assert outcome.reports[0].added == 0 + + +def test_new_key_adds_and_one_failure_keeps_the_rest(tmp_path, monkeypatch): + kb = _kb(tmp_path) + _creds(kb) + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + upsert_binding(kb, _binding(prefix="")) + good = RemoteObject(identity="s3://acme/a.txt", name="a.txt") + bad = RemoteObject(identity="s3://acme/b.txt", name="b.txt") + noise = RemoteObject(identity="s3://acme/c.bin", name="c.bin") + + def add(path: Path, kb_dir: Path, identity: str) -> str: + if identity.endswith("b.txt"): + raise ProviderError("compile failed") + return "added" + + backend = _Backend([good, bad, noise]) + outcome = sync_bindings(kb, add=add, backend_for=lambda *_args: backend) + report = outcome.reports[0] + assert report.added == 1 + assert report.failed == 1 + assert report.skipped == 1 + assert outcome.exit_code == 1 + + +def test_transient_download_retries_then_fails(tmp_path, monkeypatch): + kb = _kb(tmp_path) + _creds(kb) + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + upsert_binding(kb, _binding(prefix="")) + obj = RemoteObject(identity="s3://acme/a.txt", name="a.txt") + backend = _Backend([obj], download_failures=2) + outcome = sync_bindings( + kb, + add=lambda path, kb_dir, identity: "added", + backend_for=lambda *_args: backend, + ) + assert outcome.exit_code == 0 + assert backend.downloads == 3 + assert outcome.reports[0].added == 1 + + backend = _Backend([obj], download_failures=3) + outcome = sync_bindings( + kb, + add=lambda path, kb_dir, identity: "added", + backend_for=lambda *_args: backend, + ) + assert outcome.reports[0].failed == 1 + assert outcome.exit_code == 1 + + +def test_non_interactive_does_not_call_sign_in(): + called: list[str] = [] + with pytest.raises(ProviderError): + recover_auth(non_interactive=True, sign_in=lambda: called.append("opened")) + assert called == [] + + +def test_s3_lists_pages_and_empty_prefix_is_the_bucket_root(tmp_path): + class _Paginator: + def __init__(self) -> None: + self.kwargs: dict[str, str] = {} + + def paginate(self, **kwargs): + self.kwargs = kwargs + yield {"Contents": [{"Key": "notes.txt"}, {"Key": "dir/"}]} + yield {"Contents": [{"Key": "sub/paper.pdf"}]} + + class _Client: + def __init__(self) -> None: + self.paginator = _Paginator() + self.downloaded: list[str] = [] + + def get_paginator(self, name: str): + assert name == "list_objects_v2" + return self.paginator + + def download_file(self, bucket: str, key: str, dest: str) -> None: + self.downloaded.append(f"{bucket}/{key}") + Path(dest).write_text(key, encoding="utf-8") + + client = _Client() + binding = _binding(prefix="") + backend = S3Backend(binding, "a", "b", client=client) + found = backend.list_objects() + assert "Prefix" not in client.paginator.kwargs + assert [item.identity for item in found] == [ + "s3://acme/notes.txt", + "s3://acme/sub/paper.pdf", + ] + out = tmp_path / found[0].name + backend.download(found[0], out) + assert out.read_text(encoding="utf-8") == "notes.txt" + assert client.downloaded == ["acme/notes.txt"] + + +def test_missing_boto3_names_the_extra(): + try: + import boto3 # noqa: F401 + except ImportError: + with pytest.raises(ProviderError, match=r"openkb\[s3\]"): + _load_boto3() + else: + pytest.skip("boto3 is installed in this environment") diff --git a/tests/test_source_cli.py b/tests/test_source_cli.py new file mode 100644 index 00000000..043cae11 --- /dev/null +++ b/tests/test_source_cli.py @@ -0,0 +1,99 @@ +"""CLI and REST coverage for ``openkb source``.""" + +from __future__ import annotations + +from click.testing import CliRunner + +from openkb.cli import cli +from openkb.connectors.sync import BindingReport, SyncOutcome + + +def _kb(tmp_path): + kb = tmp_path / "kb" + (kb / ".openkb").mkdir(parents=True) + (kb / "wiki").mkdir() + (kb / ".env").write_text( + "OPENKB_S3_ACCESS_KEY_ID=AKIA_TEST\nOPENKB_S3_SECRET_ACCESS_KEY=secret-test\n", + encoding="utf-8", + ) + return kb + + +def test_source_add_requires_credentials_before_saving(tmp_path, monkeypatch): + kb = _kb(tmp_path) + (kb / ".env").write_text("", encoding="utf-8") + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + monkeypatch.chdir(kb) + result = CliRunner().invoke( + cli, + ["source", "add", "s3", "--name", "papers", "--region", "us-east-1", "--bucket", "acme"], + ) + assert result.exit_code != 0 + assert "credentials are missing" in result.output + assert ( + "connectors" not in (kb / ".openkb" / "config.yaml").read_text(encoding="utf-8") + if (kb / ".openkb" / "config.yaml").exists() + else True + ) + + +def test_source_add_list_remove(tmp_path, monkeypatch): + kb = _kb(tmp_path) + monkeypatch.delenv("OPENKB_S3_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("OPENKB_S3_SECRET_ACCESS_KEY", raising=False) + monkeypatch.chdir(kb) + runner = CliRunner() + added = runner.invoke( + cli, + [ + "source", + "add", + "s3", + "--name", + "papers", + "--region", + "us-east-1", + "--bucket", + "acme", + "--prefix", + "docs/", + "--non-interactive", + ], + ) + assert added.exit_code == 0, added.output + listed = runner.invoke(cli, ["source", "list"]) + assert "papers" in listed.output + assert "s3://acme/docs/" in listed.output + removed = runner.invoke(cli, ["source", "remove", "papers"]) + assert removed.exit_code == 0 + assert "remain in the wiki" in removed.output + + +def test_source_sync_exit_code(tmp_path, monkeypatch): + kb = _kb(tmp_path) + monkeypatch.chdir(kb) + (kb / ".openkb" / "config.yaml").write_text( + "connectors:\n- name: papers\n provider: s3\n region: us-east-1\n" + " bucket: acme\n prefix: ''\n", + encoding="utf-8", + ) + + def fake_sync(kb_dir, *, name=None, non_interactive=False, **kwargs): + assert non_interactive is True + return SyncOutcome( + reports=[ + BindingReport( + name="papers", + provider="s3", + ok=False, + failed=1, + error="one file failed", + ) + ] + ) + + monkeypatch.setattr("openkb.source_cmd.sync_bindings", fake_sync) + result = CliRunner().invoke(cli, ["source", "sync", "--non-interactive"]) + assert result.exit_code == 1 + assert "failed 1" in result.output diff --git a/uv.lock b/uv.lock index b9c12c73..76c8bee2 100644 --- a/uv.lock +++ b/uv.lock @@ -237,6 +237,34 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/39/47f9197bdd44df24d67ac8893641e16f386c984a0619ef2ee4c51fbbc019/beautifulsoup4-4.14.3-py3-none-any.whl", hash = "sha256:0918bfe44902e6ad8d57732ba310582e98da931428d231a5ecb9e7c703a735bb", size = 107721, upload-time = "2025-11-30T15:08:24.087Z" }, ] +[[package]] +name = "boto3" +version = "1.43.124" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "botocore" }, + { name = "jmespath" }, + { name = "s3transfer" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/fc/71d0afe1c7bad85524c292bcfb53d89acd5d05a944992a10066eb8e7ecd3/boto3-1.43.124.tar.gz", hash = "sha256:7f330e53e274a0b8e5982ed3dd4cec9e7e932cf48306c08db8b9e880d09c5c63", size = 112680, upload-time = "2026-11-04T21:22:36.001Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/04/7a7c75e465b063d40d63b6fc5217e312793400fb43411d3064a5ba9174cc/boto3-1.43.124-py3-none-any.whl", hash = "sha256:62edd40fdeef6523ee2f73f4ccbe754fde79200ad0763bd86f4422e8125b2735", size = 140042, upload-time = "2026-11-04T21:22:33.566Z" }, +] + +[[package]] +name = "botocore" +version = "1.43.124" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "jmespath" }, + { name = "python-dateutil" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d3/1e/c0abe23b9495414dc79e3faeb6e96e2336053fd8cae7d80cc6bbe4bd02f3/botocore-1.43.124.tar.gz", hash = "sha256:6df80235995e2926e0768e7ad270e774adcd6ff0307da7ab87ffb8c384edb578", size = 16509340, upload-time = "2026-11-04T21:22:30.041Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/84/a4/77409b07ba0b5520dbc591c33b305c5fe4c186036d881ca894a597c0dfba/botocore-1.43.124-py3-none-any.whl", hash = "sha256:4651bac68d1aef78d42c8aefb514a6cd86fda07971b8348652a9d24f21b07965", size = 16209102, upload-time = "2026-11-04T21:22:22.159Z" }, +] + [[package]] name = "certifi" version = "2026.4.22" @@ -596,7 +624,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1114,6 +1142,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c8/8d/302cb2057b7513327b4d575cff6b1d066ee6431a5357fc3f8867cd684406/jiter-0.15.0-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:54d5d6090cdc1b7c9e780dfb04949a990adb1e301a2fc0bbcee7de4638d33f9a", size = 344469, upload-time = "2026-05-19T10:09:46.864Z" }, ] +[[package]] +name = "jmespath" +version = "1.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d3/59/322338183ecda247fb5d1763a6cbe46eff7222eaeebafd9fa65d4bf5cb11/jmespath-1.1.0.tar.gz", hash = "sha256:472c87d80f36026ae83c6ddd0f1d05d4e510134ed462851fd5f754c8c3cbb88d", size = 27377, upload-time = "2026-01-22T16:35:26.279Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/14/2f/967ba146e6d58cf6a652da73885f52fc68001525b4197effc174321d70b4/jmespath-1.1.0-py3-none-any.whl", hash = "sha256:a5663118de4908c91729bea0acadca56526eb2698e83de10cd116ae0f4e97c64", size = 20419, upload-time = "2026-01-22T16:35:24.919Z" }, +] + [[package]] name = "json-repair" version = "0.59.10" @@ -1989,6 +2026,9 @@ dev = [ { name = "ruff" }, { name = "types-pyyaml" }, ] +s3 = [ + { name = "boto3" }, +] web = [ { name = "fastapi" }, { name = "python-multipart" }, @@ -1997,6 +2037,7 @@ web = [ [package.metadata] requires-dist = [ + { name = "boto3", marker = "extra == 's3'", specifier = "==1.43.124" }, { name = "click", specifier = "==8.4.0" }, { name = "fastapi", marker = "extra == 'api'" }, { name = "fastapi", marker = "extra == 'web'" }, @@ -2024,7 +2065,7 @@ requires-dist = [ { name = "uvicorn", marker = "extra == 'web'" }, { name = "watchdog", specifier = "==6.0.0" }, ] -provides-extras = ["api", "dev", "web"] +provides-extras = ["api", "dev", "s3", "web"] [[package]] name = "openpyxl" @@ -2078,10 +2119,10 @@ resolution-markers = [ "python_full_version < '3.11' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "python-dateutil", marker = "python_full_version < '3.11'" }, - { name = "pytz", marker = "python_full_version < '3.11'" }, - { name = "tzdata", marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "pytz" }, + { name = "tzdata" }, ] sdist = { url = "https://files.pythonhosted.org/packages/33/01/d40b85317f86cf08d853a4f495195c73815fdf205eef3993821720274518/pandas-2.3.3.tar.gz", hash = "sha256:e05e1af93b977f7eafa636d043f9f94c7ee3ac81af99c13508215942e64c993b", size = 4495223, upload-time = "2025-09-29T23:34:51.853Z" } wheels = [ @@ -2149,9 +2190,9 @@ resolution-markers = [ "python_full_version == '3.11.*' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "python-dateutil", marker = "python_full_version >= '3.11'" }, - { name = "tzdata", marker = "(python_full_version >= '3.11' and sys_platform == 'emscripten') or (python_full_version >= '3.11' and sys_platform == 'win32')" }, + { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "tzdata", marker = "sys_platform == 'emscripten' or sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/f8/87/4341c6252d1c47b08768c3d25ac487362bf403f0313ddae4a2a26c9b1b4c/pandas-3.0.3.tar.gz", hash = "sha256:696a4a00a2a2a35d4e5deb3fc946641b96c944f02230e4f76137fe35d806c4fc", size = 4651414, upload-time = "2026-05-11T18:54:29.21Z" } wheels = [ @@ -3174,6 +3215,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/63/6a/aca01554949f3a401991dc32fe22837baeaccb8a0d868256cbb26a029778/ruff-0.9.7-py3-none-win_arm64.whl", hash = "sha256:b075a700b2533feb7a01130ff656a4ec0d5f340bb540ad98759b8401c32c2037", size = 10177763, upload-time = "2025-02-20T13:26:48.92Z" }, ] +[[package]] +name = "s3transfer" +version = "0.19.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "botocore" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/49/c5/0a5ce655109cb96b8dfad36908a4e108d15f5dc2eff8fbc3146c4312e704/s3transfer-0.19.3.tar.gz", hash = "sha256:8326acc3d056d7f7b8a37110c7d2aea41f67e83f6ae79cd3b5fbcca34c726406", size = 166087, upload-time = "2026-10-19T19:19:49.751Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e2/a4/88f9e10498759121060b8050a1554753dfc04df39bbbcc7aad9ecc8d9440/s3transfer-0.19.3-py3-none-any.whl", hash = "sha256:24eb7535804306355e31c952a90022bf5b2f9d760f9a11062c7493a217c08553", size = 90252, upload-time = "2026-10-19T19:19:48.309Z" }, +] + [[package]] name = "shellingham" version = "1.5.4"