4c30e5eb8f
Introduces the data layer for multi-store ingestion. Documents can now live in multiple stores, each with its own Ingested/Stale/Failed state. The doc-level lifecycle (#202) becomes the aggregate over all per-store links, computed by a pure domain function. Domain - Store entity (name, slug, kind, embedder, config, is_default) - DocumentStoreLink entity with mark_ingested / mark_stale / mark_failed helpers - StoreKind and DocumentStoreLinkState enums - aggregate_lifecycle(): pure function — Failed > Stale > Ingested > fallback (the doc's pre-link Uploaded/Parsed/Chunked state) Persistence - New tables 'stores' and 'document_store_links' with the right indexes (doc_id, store_id, state) and a UNIQUE (doc, store) on the link - Default 'opensearch' store seeded idempotently in init_db, embedder pulled from DEFAULT_EMBEDDER (fallback bge-m3) - SqliteStoreRepository (find_by_slug, find_by_id, get_default, …) - SqliteDocumentStoreLinkRepository with ON CONFLICT … DO UPDATE upsert Ports - StoreRepository and DocumentStoreLinkRepository protocols added Tests - 14 new tests: seed idempotency, insert/find round-trips, UNIQUE constraint, cascade delete with the document, every link state round-trips, aggregation rule with all branches Refs #203
98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
"""Document-Store link repository — SQLite CRUD on `document_store_links`."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from domain.models import DocumentStoreLink
|
|
from domain.value_objects import DocumentStoreLinkState
|
|
from persistence.database import get_connection
|
|
|
|
|
|
def _parse_iso(value: str | None) -> datetime | None:
|
|
if value is None or value == "":
|
|
return None
|
|
parsed = datetime.fromisoformat(value)
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=UTC)
|
|
return parsed
|
|
|
|
|
|
def _row_to_link(row) -> DocumentStoreLink:
|
|
return DocumentStoreLink(
|
|
id=row["id"],
|
|
document_id=row["document_id"],
|
|
store_id=row["store_id"],
|
|
state=DocumentStoreLinkState(row["state"]),
|
|
chunkset_hash=row["chunkset_hash"],
|
|
last_push_at=_parse_iso(row["last_push_at"]),
|
|
last_run_id=row["last_run_id"],
|
|
error_message=row["error_message"],
|
|
)
|
|
|
|
|
|
class SqliteDocumentStoreLinkRepository:
|
|
"""SQLite implementation of the DocumentStoreLinkRepository port."""
|
|
|
|
async def upsert(self, link: DocumentStoreLink) -> None:
|
|
"""Insert a new link or update the existing (document_id, store_id) row."""
|
|
async with get_connection() as db:
|
|
await db.execute(
|
|
"""INSERT INTO document_store_links
|
|
(id, document_id, store_id, state, chunkset_hash,
|
|
last_push_at, last_run_id, error_message)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(document_id, store_id) DO UPDATE SET
|
|
state = excluded.state,
|
|
chunkset_hash = excluded.chunkset_hash,
|
|
last_push_at = excluded.last_push_at,
|
|
last_run_id = excluded.last_run_id,
|
|
error_message = excluded.error_message""",
|
|
(
|
|
link.id,
|
|
link.document_id,
|
|
link.store_id,
|
|
link.state.value,
|
|
link.chunkset_hash,
|
|
str(link.last_push_at) if link.last_push_at else None,
|
|
link.last_run_id,
|
|
link.error_message,
|
|
),
|
|
)
|
|
await db.commit()
|
|
|
|
async def find_for_document(self, document_id: str) -> list[DocumentStoreLink]:
|
|
async with get_connection() as db:
|
|
cursor = await db.execute(
|
|
"SELECT * FROM document_store_links WHERE document_id = ?",
|
|
(document_id,),
|
|
)
|
|
rows = await cursor.fetchall()
|
|
return [_row_to_link(r) for r in rows]
|
|
|
|
async def find_for_store(self, store_id: str) -> list[DocumentStoreLink]:
|
|
async with get_connection() as db:
|
|
cursor = await db.execute(
|
|
"SELECT * FROM document_store_links WHERE store_id = ?",
|
|
(store_id,),
|
|
)
|
|
rows = await cursor.fetchall()
|
|
return [_row_to_link(r) for r in rows]
|
|
|
|
async def find_one(self, document_id: str, store_id: str) -> DocumentStoreLink | None:
|
|
async with get_connection() as db:
|
|
cursor = await db.execute(
|
|
"SELECT * FROM document_store_links WHERE document_id = ? AND store_id = ?",
|
|
(document_id, store_id),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return _row_to_link(row) if row else None
|
|
|
|
async def delete(self, document_id: str, store_id: str) -> bool:
|
|
async with get_connection() as db:
|
|
cursor = await db.execute(
|
|
"DELETE FROM document_store_links WHERE document_id = ? AND store_id = ?",
|
|
(document_id, store_id),
|
|
)
|
|
await db.commit()
|
|
return cursor.rowcount > 0
|