Add /database and /config endpoints to the ingester API
This commit is contained in:
parent
0be2e5b24f
commit
c62166b26e
10 changed files with 236 additions and 7 deletions
|
|
@ -15,11 +15,6 @@ from haiku.rag.client.processing import (
|
||||||
)
|
)
|
||||||
from haiku.rag.client.titles import resolve_title
|
from haiku.rag.client.titles import resolve_title
|
||||||
from haiku.rag.converters import get_converter
|
from haiku.rag.converters import get_converter
|
||||||
from haiku.rag.ingester.sources import (
|
|
||||||
FetchResult,
|
|
||||||
resolve_adhoc_fetcher,
|
|
||||||
resolve_configured_source,
|
|
||||||
)
|
|
||||||
from haiku.rag.store.models.chunk import Chunk
|
from haiku.rag.store.models.chunk import Chunk
|
||||||
from haiku.rag.store.models.document import Document
|
from haiku.rag.store.models.document import Document
|
||||||
from haiku.rag.store.models.document_item import extract_items
|
from haiku.rag.store.models.document_item import extract_items
|
||||||
|
|
@ -29,7 +24,7 @@ if TYPE_CHECKING:
|
||||||
from docling_core.types.doc.document import DoclingDocument
|
from docling_core.types.doc.document import DoclingDocument
|
||||||
|
|
||||||
from haiku.rag.client import HaikuRAG
|
from haiku.rag.client import HaikuRAG
|
||||||
from haiku.rag.ingester.sources.base import Source
|
from haiku.rag.ingester.sources.base import FetchResult, Source
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
@ -330,7 +325,7 @@ async def _refresh_doc_metadata(
|
||||||
|
|
||||||
async def _ingest_fetch_result(
|
async def _ingest_fetch_result(
|
||||||
client: "HaikuRAG",
|
client: "HaikuRAG",
|
||||||
result: FetchResult,
|
result: "FetchResult",
|
||||||
*,
|
*,
|
||||||
title: str | None,
|
title: str | None,
|
||||||
user_metadata: dict,
|
user_metadata: dict,
|
||||||
|
|
@ -495,6 +490,8 @@ async def _reconcile_pdf_attachments(
|
||||||
):
|
):
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
from haiku.rag.ingester.sources.base import FetchResult
|
||||||
|
|
||||||
child_fr = FetchResult(
|
child_fr = FetchResult(
|
||||||
uri=child_uri,
|
uri=child_uri,
|
||||||
body=data,
|
body=data,
|
||||||
|
|
@ -601,6 +598,11 @@ async def create_document_from_source(
|
||||||
# renamed/removed source surfaces as a DLQ instead of silently dropping
|
# renamed/removed source surfaces as a DLQ instead of silently dropping
|
||||||
# credentials. Ad-hoc CLI calls (no source_id) fall back to scheme-based
|
# credentials. Ad-hoc CLI calls (no source_id) fall back to scheme-based
|
||||||
# adapters when no configured source matches.
|
# adapters when no configured source matches.
|
||||||
|
from haiku.rag.ingester.sources import (
|
||||||
|
resolve_adhoc_fetcher,
|
||||||
|
resolve_configured_source,
|
||||||
|
)
|
||||||
|
|
||||||
if source_id is not None:
|
if source_id is not None:
|
||||||
fetcher = resolve_configured_source(source_str, source_id, sources)
|
fetcher = resolve_configured_source(source_str, source_id, sources)
|
||||||
else:
|
else:
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ from haiku.rag.config.loader import (
|
||||||
find_config_file,
|
find_config_file,
|
||||||
generate_default_config,
|
generate_default_config,
|
||||||
load_yaml_config,
|
load_yaml_config,
|
||||||
|
redact_secrets,
|
||||||
)
|
)
|
||||||
from haiku.rag.config.models import (
|
from haiku.rag.config.models import (
|
||||||
APIConfig,
|
APIConfig,
|
||||||
|
|
@ -60,6 +61,7 @@ __all__ = [
|
||||||
"generate_default_config",
|
"generate_default_config",
|
||||||
"get_config",
|
"get_config",
|
||||||
"load_yaml_config",
|
"load_yaml_config",
|
||||||
|
"redact_secrets",
|
||||||
"set_config",
|
"set_config",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
|
|
@ -56,3 +57,32 @@ def generate_default_config() -> dict:
|
||||||
|
|
||||||
default_config = AppConfig()
|
default_config = AppConfig()
|
||||||
return default_config.model_dump(mode="json", exclude_none=False)
|
return default_config.model_dump(mode="json", exclude_none=False)
|
||||||
|
|
||||||
|
|
||||||
|
_SECRET_KEY_HINTS = ("key", "password", "token", "secret")
|
||||||
|
|
||||||
|
|
||||||
|
def _is_secret_key(key: str) -> bool:
|
||||||
|
lowered = key.lower()
|
||||||
|
return any(hint in lowered for hint in _SECRET_KEY_HINTS)
|
||||||
|
|
||||||
|
|
||||||
|
def redact_secrets(data: Any) -> Any:
|
||||||
|
"""Recursively mask secret-bearing values in a config dump. Any scalar
|
||||||
|
whose key contains key/password/token/secret becomes "***" when set or
|
||||||
|
None when unset; everything else is preserved."""
|
||||||
|
if isinstance(data, dict):
|
||||||
|
result = {}
|
||||||
|
for key, value in data.items():
|
||||||
|
if (
|
||||||
|
isinstance(key, str)
|
||||||
|
and _is_secret_key(key)
|
||||||
|
and not isinstance(value, (dict, list))
|
||||||
|
):
|
||||||
|
result[key] = "***" if value else None
|
||||||
|
else:
|
||||||
|
result[key] = redact_secrets(value)
|
||||||
|
return result
|
||||||
|
if isinstance(data, list):
|
||||||
|
return [redact_secrets(item) for item in data]
|
||||||
|
return data
|
||||||
|
|
|
||||||
17
haiku_rag_slim/haiku/rag/ingester/api/routes/config.py
Normal file
17
haiku_rag_slim/haiku/rag/ingester/api/routes/config.py
Normal file
|
|
@ -0,0 +1,17 @@
|
||||||
|
import yaml
|
||||||
|
from fastapi import APIRouter, Depends
|
||||||
|
|
||||||
|
from haiku.rag.config import redact_secrets
|
||||||
|
from haiku.rag.ingester.api.schemas import ConfigResponse
|
||||||
|
from haiku.rag.ingester.api.server import APIState, get_state
|
||||||
|
|
||||||
|
router = APIRouter(tags=["config"])
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/config", response_model=ConfigResponse)
|
||||||
|
async def config(state: APIState = Depends(get_state)) -> ConfigResponse:
|
||||||
|
"""The full effective configuration (defaults filled in, not the on-disk
|
||||||
|
file) as YAML, with secret-bearing values redacted."""
|
||||||
|
data = redact_secrets(state.config.model_dump(mode="json", exclude_none=False))
|
||||||
|
text = yaml.dump(data, sort_keys=False, default_flow_style=False)
|
||||||
|
return ConfigResponse(yaml=text)
|
||||||
22
haiku_rag_slim/haiku/rag/ingester/api/routes/database.py
Normal file
22
haiku_rag_slim/haiku/rag/ingester/api/routes/database.py
Normal file
|
|
@ -0,0 +1,22 @@
|
||||||
|
from fastapi import APIRouter, Depends, HTTPException, status
|
||||||
|
|
||||||
|
from haiku.rag.ingester.api.server import APIState, get_state
|
||||||
|
from haiku.rag.store.engine import DatabaseInfo, gather_database_info
|
||||||
|
|
||||||
|
router = APIRouter(tags=["database"])
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/database", response_model=DatabaseInfo)
|
||||||
|
async def database(state: APIState = Depends(get_state)) -> DatabaseInfo:
|
||||||
|
"""Read-only snapshot of the LanceDB target — stored version, embeddings,
|
||||||
|
per-table counts/sizes, vector index status, pending migrations and
|
||||||
|
package versions. The same data the `haiku-rag info` command prints.
|
||||||
|
|
||||||
|
Opens a fresh read-only connection per call; not cached and not part of
|
||||||
|
the dashboard's polling loop."""
|
||||||
|
if state.db_path is None:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||||
|
detail="database path not configured",
|
||||||
|
)
|
||||||
|
return await gather_database_info(state.config, state.db_path)
|
||||||
|
|
@ -70,6 +70,13 @@ class WorkerStats(BaseModel):
|
||||||
total: int
|
total: int
|
||||||
|
|
||||||
|
|
||||||
|
class ConfigResponse(BaseModel):
|
||||||
|
"""The full effective config (defaults filled in) as YAML, with secrets
|
||||||
|
redacted."""
|
||||||
|
|
||||||
|
yaml: str
|
||||||
|
|
||||||
|
|
||||||
class StatsResponse(BaseModel):
|
class StatsResponse(BaseModel):
|
||||||
"""Aggregated counters and per-source breakdowns that drive the dashboard.
|
"""Aggregated counters and per-source breakdowns that drive the dashboard.
|
||||||
Cheap to compute (all SQL aggregations against the queue file)."""
|
Cheap to compute (all SQL aggregations against the queue file)."""
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from fastapi import Depends, FastAPI, Request
|
from fastapi import Depends, FastAPI, Request
|
||||||
|
|
@ -22,6 +23,7 @@ class APIState:
|
||||||
sync_repo: "SyncStateRepo"
|
sync_repo: "SyncStateRepo"
|
||||||
pool: "WorkerPool | None" = None
|
pool: "WorkerPool | None" = None
|
||||||
pollers: "PollerManager | None" = None
|
pollers: "PollerManager | None" = None
|
||||||
|
db_path: Path | None = None
|
||||||
|
|
||||||
|
|
||||||
def get_state(request: Request) -> APIState:
|
def get_state(request: Request) -> APIState:
|
||||||
|
|
@ -34,8 +36,12 @@ def build_app(
|
||||||
auth_token: str | None = None,
|
auth_token: str | None = None,
|
||||||
) -> FastAPI:
|
) -> FastAPI:
|
||||||
"""Construct the ingester's FastAPI control plane."""
|
"""Construct the ingester's FastAPI control plane."""
|
||||||
|
from haiku.rag.ingester.api.routes import (
|
||||||
|
config as config_route,
|
||||||
|
)
|
||||||
from haiku.rag.ingester.api.routes import (
|
from haiku.rag.ingester.api.routes import (
|
||||||
dashboard,
|
dashboard,
|
||||||
|
database,
|
||||||
dlq,
|
dlq,
|
||||||
health,
|
health,
|
||||||
jobs,
|
jobs,
|
||||||
|
|
@ -63,5 +69,7 @@ def build_app(
|
||||||
app.include_router(dlq.router, dependencies=auth_dep)
|
app.include_router(dlq.router, dependencies=auth_dep)
|
||||||
app.include_router(stats.router, dependencies=auth_dep)
|
app.include_router(stats.router, dependencies=auth_dep)
|
||||||
app.include_router(providers.router, dependencies=auth_dep)
|
app.include_router(providers.router, dependencies=auth_dep)
|
||||||
|
app.include_router(database.router, dependencies=auth_dep)
|
||||||
|
app.include_router(config_route.router, dependencies=auth_dep)
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
|
|
||||||
|
|
@ -244,6 +244,7 @@ class IngesterApp:
|
||||||
sync_repo=self._sync,
|
sync_repo=self._sync,
|
||||||
pool=self._pool,
|
pool=self._pool,
|
||||||
pollers=self._pollers,
|
pollers=self._pollers,
|
||||||
|
db_path=self._db_path,
|
||||||
)
|
)
|
||||||
if ingester_cfg.api.auth_token is None:
|
if ingester_cfg.api.auth_token is None:
|
||||||
logger.warning("API auth_token is unset — control plane is unauthenticated")
|
logger.warning("API auth_token is unset — control plane is unauthenticated")
|
||||||
|
|
|
||||||
|
|
@ -633,3 +633,119 @@ async def test_dashboard_served_unauthenticated(state):
|
||||||
assert "/jobs?status=claimed" in body
|
assert "/jobs?status=claimed" in body
|
||||||
# Op badge helper is present so DELETE rows render distinctly.
|
# Op badge helper is present so DELETE rows render distinctly.
|
||||||
assert "opBadge" in body
|
assert "opBadge" in body
|
||||||
|
|
||||||
|
|
||||||
|
# --- config ---
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_config_returns_full_yaml_with_redacted_secrets(jobs, sync):
|
||||||
|
from haiku.rag.config import APIConfig, IngesterConfig
|
||||||
|
|
||||||
|
config = AppConfig(ingester=IngesterConfig(api=APIConfig(auth_token="supersecret")))
|
||||||
|
state = APIState(config=config, job_repo=jobs, sync_repo=sync)
|
||||||
|
async with _client(state) as client:
|
||||||
|
resp = await client.get("/config")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
text = resp.json()["yaml"]
|
||||||
|
# Full effective config: a default section the user never wrote is present.
|
||||||
|
assert "embeddings:" in text
|
||||||
|
assert "processing:" in text
|
||||||
|
# Secret is masked, not echoed.
|
||||||
|
assert "supersecret" not in text
|
||||||
|
assert "auth_token: '***'" in text
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_config_requires_auth(state):
|
||||||
|
async with _client(state, auth_token="secret") as client:
|
||||||
|
resp = await client.get("/config")
|
||||||
|
assert resp.status_code == 401
|
||||||
|
|
||||||
|
|
||||||
|
# --- database ---
|
||||||
|
|
||||||
|
|
||||||
|
async def _seed_lancedb(path):
|
||||||
|
"""Create a minimal LanceDB with settings/documents/chunks tables so
|
||||||
|
gather_database_info has something real to report."""
|
||||||
|
import json
|
||||||
|
|
||||||
|
import lancedb
|
||||||
|
from lancedb.pydantic import LanceModel, Vector
|
||||||
|
from pydantic import Field
|
||||||
|
|
||||||
|
class SettingsRecord(LanceModel):
|
||||||
|
id: str = Field(default="settings")
|
||||||
|
settings: str = Field(default="{}")
|
||||||
|
|
||||||
|
class DocumentRecord(LanceModel):
|
||||||
|
id: str
|
||||||
|
content: str
|
||||||
|
|
||||||
|
class ChunkRecord(LanceModel):
|
||||||
|
id: str
|
||||||
|
document_id: str
|
||||||
|
content: str
|
||||||
|
vector: Vector(3) # type: ignore
|
||||||
|
|
||||||
|
db = await lancedb.connect_async(str(path))
|
||||||
|
settings_tbl = await db.create_table("settings", schema=SettingsRecord)
|
||||||
|
docs_tbl = await db.create_table("documents", schema=DocumentRecord)
|
||||||
|
chunks_tbl = await db.create_table("chunks", schema=ChunkRecord)
|
||||||
|
await settings_tbl.add(
|
||||||
|
[
|
||||||
|
SettingsRecord(
|
||||||
|
settings=json.dumps(
|
||||||
|
{
|
||||||
|
"version": "1.2.3",
|
||||||
|
"embeddings": {
|
||||||
|
"model": {
|
||||||
|
"provider": "openai",
|
||||||
|
"name": "text-embedding-3-small",
|
||||||
|
"vector_dim": 3,
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
)
|
||||||
|
]
|
||||||
|
)
|
||||||
|
await docs_tbl.add([DocumentRecord(id="doc-1", content="hello")])
|
||||||
|
await chunks_tbl.add(
|
||||||
|
[ChunkRecord(id="c1", document_id="doc-1", content="c", vector=[0.1, 0.2, 0.3])]
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_database_reports_info(tmp_path, jobs, sync):
|
||||||
|
db_path = tmp_path / "docs.lancedb"
|
||||||
|
await _seed_lancedb(db_path)
|
||||||
|
state = APIState(config=AppConfig(), job_repo=jobs, sync_repo=sync, db_path=db_path)
|
||||||
|
async with _client(state) as client:
|
||||||
|
resp = await client.get("/database")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
body = resp.json()
|
||||||
|
assert body["exists"] is True
|
||||||
|
assert body["stored_version"] == "1.2.3"
|
||||||
|
assert body["embeddings"]["provider"] == "openai"
|
||||||
|
assert body["embeddings"]["vector_dim"] == 3
|
||||||
|
tables = {t["name"]: t for t in body["tables"]}
|
||||||
|
assert tables["documents"]["num_rows"] == 1
|
||||||
|
assert tables["chunks"]["num_rows"] == 1
|
||||||
|
assert tables["document_items"]["exists"] is False
|
||||||
|
assert body["vector_index"]["exists"] is False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_database_503_when_db_path_unset(state):
|
||||||
|
async with _client(state) as client:
|
||||||
|
resp = await client.get("/database")
|
||||||
|
assert resp.status_code == 503
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_database_requires_auth(state):
|
||||||
|
async with _client(state, auth_token="secret") as client:
|
||||||
|
resp = await client.get("/database")
|
||||||
|
assert resp.status_code == 401
|
||||||
|
|
|
||||||
|
|
@ -350,3 +350,27 @@ analysis:
|
||||||
assert cfg.analysis.model.name == "analysis-model"
|
assert cfg.analysis.model.name == "analysis-model"
|
||||||
resolved = cfg.analysis.model or cfg.qa.model
|
resolved = cfg.analysis.model or cfg.qa.model
|
||||||
assert resolved.name == "analysis-model"
|
assert resolved.name == "analysis-model"
|
||||||
|
|
||||||
|
|
||||||
|
def test_redact_secrets_masks_nested_secret_keys():
|
||||||
|
from haiku.rag.config.loader import redact_secrets
|
||||||
|
|
||||||
|
data = {
|
||||||
|
"api_key": "sk-123",
|
||||||
|
"name": "ollama",
|
||||||
|
"ingester": {"api": {"auth_token": "secret", "host": "0.0.0.0"}},
|
||||||
|
"missing_token": None,
|
||||||
|
"sources": [{"password": "pw", "url": "http://x"}],
|
||||||
|
"storage_options": {"aws_secret_access_key": "abc"},
|
||||||
|
}
|
||||||
|
|
||||||
|
redacted = redact_secrets(data)
|
||||||
|
|
||||||
|
assert redacted["api_key"] == "***"
|
||||||
|
assert redacted["name"] == "ollama"
|
||||||
|
assert redacted["ingester"]["api"]["auth_token"] == "***"
|
||||||
|
assert redacted["ingester"]["api"]["host"] == "0.0.0.0"
|
||||||
|
assert redacted["missing_token"] is None
|
||||||
|
assert redacted["sources"][0]["password"] == "***"
|
||||||
|
assert redacted["sources"][0]["url"] == "http://x"
|
||||||
|
assert redacted["storage_options"]["aws_secret_access_key"] == "***"
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue