| """CatalogReader β loads + filters catalog by source_hint. |
| |
| For typical users (β€50 tables), returns the FULL catalog with no slicing. |
| Catalog-level search is added later if catalog grows past the limit. |
| """ |
|
|
| from datetime import UTC, datetime |
| from typing import Literal |
|
|
| from src.middlewares.logging import get_logger |
|
|
| from .models import Catalog, Source |
| from .store import CatalogStore |
|
|
| logger = get_logger("catalog_reader") |
|
|
| SourceHint = Literal["chat", "unstructured", "structured"] |
|
|
|
|
| def _filter_sources(catalog: Catalog, source_hint: SourceHint) -> Catalog: |
| """Return a copy of `catalog` keeping only the sources matching `source_hint`.""" |
| filtered: list[Source] |
| if source_hint == "chat": |
| filtered = [] |
| elif source_hint == "structured": |
| filtered = [s for s in catalog.sources if s.source_type in {"schema", "tabular"}] |
| else: |
| filtered = [s for s in catalog.sources if s.source_type == "unstructured"] |
| return catalog.model_copy(update={"sources": filtered}) |
|
|
|
|
| class CatalogReader: |
| """Loads the user's catalog and filters by source_hint. |
| |
| On miss, returns an empty Catalog (never raises) β query path is |
| responsible for handling "no data registered yet" gracefully. |
| Returned Catalog is always a copy; the underlying stored catalog |
| is never mutated. |
| """ |
|
|
| def __init__(self, store: CatalogStore) -> None: |
| self._store = store |
|
|
| async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: |
| catalog = await self._store.get(user_id) |
| if catalog is None: |
| return Catalog(user_id=user_id, generated_at=datetime.now(UTC)) |
|
|
| return _filter_sources(catalog, source_hint) |
|
|
|
|
| class MemoizingCatalogReader(CatalogReader): |
| """Request-scoped CatalogReader that caches each ``read`` by source_hint. |
| |
| One per request. The same per-user catalog is otherwise fetched from the |
| catalog DB 4-5x during a single slow-path run (planner load, then |
| check_data's structured read + check_knowledge's unstructured read, then |
| retrieve_data's structured read). Wrapping the base reader collapses those |
| to one round-trip |
| per distinct source_hint and pins a single consistent snapshot for the whole |
| request (plan-time and execution-time catalogs can no longer diverge). |
| """ |
|
|
| def __init__(self, inner: CatalogReader) -> None: |
| |
| |
| |
| super().__init__(getattr(inner, "_store", None)) |
| self._inner = inner |
| self._cache: dict[SourceHint, Catalog] = {} |
|
|
| async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: |
| cached = self._cache.get(source_hint) |
| if cached is None: |
| cached = await self._inner.read(user_id, source_hint) |
| self._cache[source_hint] = cached |
| return cached |
|
|
|
|
| class AnalysisScopedCatalogReader(CatalogReader): |
| """Reads the analysis-scope catalog, falling back to the user-scope reader. |
| |
| Used by the `check` skill so "what data do I have" inside a room reflects |
| that analysis's bound sources β structured AND documents β with their real |
| names. A database shows as "xl test" (analysis-scope) instead of the |
| auto-generated `postgres_<hash>` placeholder, and documents show at all |
| (the user-scope catalog holds no `unstructured` sources, so reading them from |
| user-scope always came back empty). |
| |
| Fallback rule (tightened 2026-07-13): the user-scope reader is used ONLY when |
| there is no `analysis_id` at all (a legacy room with no analysis concept). |
| When an `analysis_id` IS present but its catalog row is missing (legacy |
| analysis created before catalog materialization, or Go hasn't rebuilt the |
| binding yet) or the read fails, this returns an EMPTY catalog β NOT the |
| user-scope catalog. Previously it degraded to user-scope, which silently |
| surfaced sources that are NOT bound to this analysis as if they were (e.g. a |
| room bound to a CSV answered "check" with the account's unrelated XLSX). An |
| empty result reads correctly as "nothing bound yet β re-save / rebuild the |
| binding to materialize the analysis catalog". Every outcome is logged so a |
| miss is diagnosable instead of silent. |
| """ |
|
|
| def __init__(self, inner: CatalogReader, analysis_id: str | None) -> None: |
| |
| |
| super().__init__(inner._store) |
| self._inner = inner |
| self._analysis_id = analysis_id |
|
|
| async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: |
| |
| |
| if not self._analysis_id: |
| return await self._inner.read(user_id, source_hint) |
|
|
| |
| |
| |
| try: |
| catalog = await self._store.get_by_analysis(self._analysis_id) |
| except Exception as e: |
| logger.warning( |
| "analysis catalog read failed β returning empty", |
| analysis_id=self._analysis_id, |
| error=str(e), |
| ) |
| catalog = None |
|
|
| if catalog is not None: |
| logger.info( |
| "analysis catalog hit", |
| analysis_id=self._analysis_id, |
| sources=len(catalog.sources), |
| ) |
| return _filter_sources(catalog, source_hint) |
|
|
| |
| |
| |
| logger.info( |
| "analysis catalog miss β returning empty (no user-scope fallback)", |
| analysis_id=self._analysis_id, |
| ) |
| return Catalog(user_id=user_id, generated_at=datetime.now(UTC)) |
|
|