feat(#251): StoreService — CRUD orchestration + per-kind config validation
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
This commit is contained in:
Pier-Jean Malandrino
parent
ce140bab43
commit
9d2e9ab0f4
2 changed files with 495 additions and 0 deletions
233
document-parser/services/store_service.py
Normal file
233
document-parser/services/store_service.py
Normal file
|
|
@ -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)
|
||||
262
document-parser/tests/test_store_service.py
Normal file
262
document-parser/tests/test_store_service.py
Normal file
|
|
@ -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")
|
||||
Loading…
Reference in a new issue