From 9d2e9ab0f4fd70500a9bac90bf327ee929874f12 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Mon, 4 May 2026 12:19:43 +0200 Subject: [PATCH] =?UTF-8?q?feat(#251):=20StoreService=20=E2=80=94=20CRUD?= =?UTF-8?q?=20orchestration=20+=20per-kind=20config=20validation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Couche service entre l'API et les repos SQLite : - list_stores avec enrichissement document_count via document_store_links - get/create/update/delete avec validations - Slug normalisé lowercase, motif kebab-case strict - Unicité name + slug (409 sur collision) - Single-default invariant via clear_default_except - Validation config par kind (OpenSearch require index_name) — extensible - Delete refusé sur le store seedé "default" et sur tout store avec liens - Erreurs typées (Validation/NotFound/Conflict) avec hint http_status --- document-parser/services/store_service.py | 233 +++++++++++++++++ document-parser/tests/test_store_service.py | 262 ++++++++++++++++++++ 2 files changed, 495 insertions(+) create mode 100644 document-parser/services/store_service.py create mode 100644 document-parser/tests/test_store_service.py diff --git a/document-parser/services/store_service.py b/document-parser/services/store_service.py new file mode 100644 index 0000000..e94f77f --- /dev/null +++ b/document-parser/services/store_service.py @@ -0,0 +1,233 @@ +"""Store service — CRUD orchestration for ingestion targets (#251). + +Sits between the API layer and the SQLite repositories. Owns: +- input validation (per-kind config schema, slug shape, name uniqueness) +- single-default invariant (only one Store can be `is_default = True`) +- delete safety (refuse on seeded `default` slug or non-empty links) +- list enrichment (per-store document counts read from + `document_store_links`) +""" + +from __future__ import annotations + +import logging +import re +import uuid +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from domain.models import Store +from domain.value_objects import StoreKind + +if TYPE_CHECKING: + from persistence.document_store_link_repo import SqliteDocumentStoreLinkRepository + from persistence.store_repo import SqliteStoreRepository + +logger = logging.getLogger(__name__) + + +_SLUG_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]*[a-z0-9]$|^[a-z0-9]$") + + +class StoreServiceError(Exception): + """Base service error. Carries an `http_status` hint for the API layer.""" + + http_status: int = 400 + + def __init__(self, message: str, *, http_status: int | None = None): + super().__init__(message) + if http_status is not None: + self.http_status = http_status + + +class StoreNotFoundError(StoreServiceError): + http_status = 404 + + +class StoreConflictError(StoreServiceError): + http_status = 409 + + +class StoreValidationError(StoreServiceError): + http_status = 422 + + +@dataclass +class StoreInfoView: + """Read model for `GET /api/stores`. Mirrors the frontend `StoreInfo`.""" + + name: str + slug: str + kind: str + embedder: str + is_default: bool + document_count: int + chunk_count: int + connected: bool + error_message: str | None = None + + +def _validate_slug(slug: str) -> None: + if not slug or not _SLUG_PATTERN.match(slug): + raise StoreValidationError( + "slug must be lowercase alphanumeric with optional dashes (e.g. 'rh-corpus-v3')" + ) + + +def _validate_config_for_kind(kind: StoreKind, config: dict) -> None: + """Per-kind config schema. Today only OpenSearch is wired; new kinds plug + in here without touching the rest of the pipeline.""" + if kind is StoreKind.OPENSEARCH: + index_name = config.get("index_name") or config.get("indexName") + if not isinstance(index_name, str) or not index_name.strip(): + raise StoreValidationError("OpenSearch config requires a non-empty 'index_name'") + # Future: add LlamaIndex / LangChain / pgvector branches here. + + +class StoreService: + """Orchestrates store CRUD on top of SQLite repositories.""" + + def __init__( + self, + store_repo: SqliteStoreRepository, + link_repo: SqliteDocumentStoreLinkRepository, + ): + self._stores = store_repo + self._links = link_repo + + async def list_stores(self) -> list[StoreInfoView]: + stores = await self._stores.find_all() + views: list[StoreInfoView] = [] + for store in stores: + links = await self._links.find_for_store(store.id) + views.append( + StoreInfoView( + name=store.name, + slug=store.slug, + kind=store.kind.value, + embedder=store.embedder, + is_default=store.is_default, + document_count=len(links), + chunk_count=0, + connected=True, + ) + ) + return views + + async def get_by_slug(self, slug: str) -> Store: + store = await self._stores.find_by_slug(slug) + if store is None: + raise StoreNotFoundError(f"Store '{slug}' not found") + return store + + async def create_store( + self, + *, + name: str, + slug: str, + kind: StoreKind, + embedder: str, + config: dict, + is_default: bool = False, + ) -> Store: + name = (name or "").strip() + slug = (slug or "").strip().lower() + embedder = (embedder or "").strip() + + if not name: + raise StoreValidationError("name is required") + if not embedder: + raise StoreValidationError("embedder is required") + _validate_slug(slug) + _validate_config_for_kind(kind, config) + + if await self._stores.find_by_slug(slug) is not None: + raise StoreConflictError(f"slug '{slug}' is already in use") + if await self._stores.find_by_name(name) is not None: + raise StoreConflictError(f"name '{name}' is already in use") + + store = Store( + id=str(uuid.uuid4()), + name=name, + slug=slug, + kind=kind, + embedder=embedder, + config=config, + is_default=is_default, + ) + await self._stores.insert(store) + if is_default: + await self._stores.clear_default_except(store.id) + return store + + async def update_store( + self, + slug: str, + *, + name: str | None = None, + new_slug: str | None = None, + kind: StoreKind | None = None, + embedder: str | None = None, + config: dict | None = None, + is_default: bool | None = None, + ) -> Store: + store = await self.get_by_slug(slug) + + if name is not None: + name = name.strip() + if not name: + raise StoreValidationError("name cannot be empty") + other = await self._stores.find_by_name(name) + if other is not None and other.id != store.id: + raise StoreConflictError(f"name '{name}' is already in use") + store.name = name + + if new_slug is not None: + new_slug = new_slug.strip().lower() + _validate_slug(new_slug) + if new_slug != store.slug: + other = await self._stores.find_by_slug(new_slug) + if other is not None and other.id != store.id: + raise StoreConflictError(f"slug '{new_slug}' is already in use") + store.slug = new_slug + + if kind is not None: + store.kind = kind + + if embedder is not None: + embedder = embedder.strip() + if not embedder: + raise StoreValidationError("embedder cannot be empty") + store.embedder = embedder + + if config is not None: + store.config = config + + # Validate the (kind, config) pair as a whole — even when only one of + # the two changed, the combination must still satisfy the schema. + _validate_config_for_kind(store.kind, store.config) + + promote_default = False + if is_default is not None: + store.is_default = is_default + promote_default = is_default + + await self._stores.update(store) + if promote_default: + await self._stores.clear_default_except(store.id) + return store + + async def delete_store(self, slug: str) -> None: + store = await self.get_by_slug(slug) + if store.slug == "default": + raise StoreConflictError( + "the seeded 'default' store cannot be deleted", + http_status=409, + ) + links = await self._links.find_for_store(store.id) + if links: + raise StoreConflictError( + f"store '{slug}' has {len(links)} linked document(s); remove the documents first", + http_status=409, + ) + await self._stores.delete(store.id) diff --git a/document-parser/tests/test_store_service.py b/document-parser/tests/test_store_service.py new file mode 100644 index 0000000..31cab80 --- /dev/null +++ b/document-parser/tests/test_store_service.py @@ -0,0 +1,262 @@ +"""Tests for StoreService — CRUD + validations (#251).""" + +from __future__ import annotations + +import pytest + +from domain.models import Document, DocumentStoreLink +from domain.value_objects import DocumentStoreLinkState, StoreKind +from persistence.database import init_db +from persistence.document_repo import SqliteDocumentRepository +from persistence.document_store_link_repo import SqliteDocumentStoreLinkRepository +from persistence.store_repo import SqliteStoreRepository +from services.store_service import ( + StoreConflictError, + StoreNotFoundError, + StoreService, + StoreValidationError, +) + + +@pytest.fixture(autouse=True) +async def setup_db(monkeypatch, tmp_path): + db_path = str(tmp_path / "test.db") + monkeypatch.setattr("persistence.database.DB_PATH", db_path) + await init_db() + yield + + +@pytest.fixture +def service(): + return StoreService(SqliteStoreRepository(), SqliteDocumentStoreLinkRepository()) + + +VALID_OS_CONFIG = {"index_name": "rh-corpus-v3"} + + +class TestCreate: + async def test_create_ok(self, service): + store = await service.create_store( + name="rh-corpus", + slug="rh-corpus", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + assert store.id + assert store.slug == "rh-corpus" + + async def test_slug_normalized_to_lowercase(self, service): + store = await service.create_store( + name="RH", + slug="RH-Corpus", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + assert store.slug == "rh-corpus" + + async def test_create_rejects_duplicate_slug(self, service): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + with pytest.raises(StoreConflictError): + await service.create_store( + name="rh-2", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + + async def test_create_rejects_duplicate_name(self, service): + await service.create_store( + name="rh", + slug="rh-1", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + with pytest.raises(StoreConflictError): + await service.create_store( + name="rh", + slug="rh-2", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + + async def test_create_rejects_bad_slug(self, service): + with pytest.raises(StoreValidationError): + await service.create_store( + name="rh", + slug="RH Corpus", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + + async def test_create_rejects_missing_index_name(self, service): + with pytest.raises(StoreValidationError): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={}, + ) + + async def test_create_default_clears_others(self, service): + # The seeded 'default' store starts as is_default=True. + new_default = await service.create_store( + name="new", + slug="new", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={"index_name": "new"}, + is_default=True, + ) + all_stores = await service.list_stores() + defaults = [s for s in all_stores if s.is_default] + assert len(defaults) == 1 + assert defaults[0].slug == new_default.slug + + +class TestRead: + async def test_get_by_slug_404(self, service): + with pytest.raises(StoreNotFoundError): + await service.get_by_slug("missing") + + async def test_list_includes_seeded_default(self, service): + stores = await service.list_stores() + slugs = [s.slug for s in stores] + assert "default" in slugs + + async def test_list_counts_linked_documents(self, service): + # Seed a doc + a link to the seeded default store. + doc_repo = SqliteDocumentRepository() + await doc_repo.insert(Document(id="d-1", filename="t.pdf", storage_path="/tmp/t.pdf")) + await SqliteDocumentStoreLinkRepository().upsert( + DocumentStoreLink( + id="l-1", + document_id="d-1", + store_id="default", + state=DocumentStoreLinkState.INGESTED, + ) + ) + views = await service.list_stores() + default_view = next(v for v in views if v.slug == "default") + assert default_view.document_count == 1 + + +class TestUpdate: + async def test_partial_update(self, service): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + updated = await service.update_store("rh", embedder="bge-large") + assert updated.embedder == "bge-large" + # Other fields untouched. + assert updated.name == "rh" + + async def test_update_rename_slug(self, service): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + updated = await service.update_store("rh", new_slug="rh-v2") + assert updated.slug == "rh-v2" + with pytest.raises(StoreNotFoundError): + await service.get_by_slug("rh") + + async def test_update_rejects_slug_collision(self, service): + await service.create_store( + name="a", + slug="a", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={"index_name": "a"}, + ) + await service.create_store( + name="b", + slug="b", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={"index_name": "b"}, + ) + with pytest.raises(StoreConflictError): + await service.update_store("a", new_slug="b") + + async def test_update_promote_default_demotes_others(self, service): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + await service.update_store("rh", is_default=True) + defaults = [s for s in await service.list_stores() if s.is_default] + assert len(defaults) == 1 + assert defaults[0].slug == "rh" + + async def test_update_rejects_invalid_config(self, service): + await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + with pytest.raises(StoreValidationError): + await service.update_store("rh", config={}) + + +class TestDelete: + async def test_delete_ok(self, service): + await service.create_store( + name="tmp", + slug="tmp", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={"index_name": "tmp"}, + ) + await service.delete_store("tmp") + with pytest.raises(StoreNotFoundError): + await service.get_by_slug("tmp") + + async def test_delete_seeded_default_refused(self, service): + with pytest.raises(StoreConflictError): + await service.delete_store("default") + + async def test_delete_with_links_refused(self, service): + store = await service.create_store( + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config=VALID_OS_CONFIG, + ) + doc_repo = SqliteDocumentRepository() + await doc_repo.insert(Document(id="d-1", filename="t.pdf", storage_path="/tmp/t.pdf")) + await SqliteDocumentStoreLinkRepository().upsert( + DocumentStoreLink( + id="l-1", + document_id="d-1", + store_id=store.id, + state=DocumentStoreLinkState.INGESTED, + ) + ) + with pytest.raises(StoreConflictError): + await service.delete_store("rh")