fix: address xhigh review findings for PR #272

- cloud: _get_folder_id raises CollectionNotFoundError when the folder is
  gone instead of silently falling back to the account-wide space; delete
  stays idempotent
- cloud: CloudBackend honors a base_url override (PageIndexClient.BASE_URL
  previously only reached the legacy API)
- cloud: query_stream runs doc-id listing and thread join off the event
  loop (asyncio.to_thread)
- cloud_api: non-streaming chat_completions gets a 300s read timeout
- local: unexplained IntegrityError is wrapped as CollectionNotFoundError/
  IndexingError instead of leaking raw sqlite3 errors
- index: check_title_appearance rejects below-range physical_index instead
  of wrapping to the last page; page_index_main coerces legacy 'yes'/'no'
  string flags
- config: correct max_concurrency doc and warn when a per-index value
  exceeds the process ceiling
- compat: restore pageindex.utils config alias; add legacy names back to
  __all__
This commit is contained in:
Ray 2026-07-16 01:31:57 +08:00
parent fec44e16e4
commit c8aa83ac85
10 changed files with 360 additions and 14 deletions

View file

@ -64,6 +64,11 @@ __all__ = [
# Legacy top-level exports (pre-SDK API), kept so `from pageindex import *`
# still binds them.
"page_index",
"page_index_main",
"tree_parser",
"ConfigLoader",
"llm_completion",
"llm_acompletion",
"md_to_tree",
"get_document",
"get_document_structure",

View file

@ -14,7 +14,8 @@ import requests
from typing import AsyncIterator
from ..cloud_api import API_BASE # single source of truth for the cloud base URL
from ..errors import AUTH_HINT, CloudAPIError, DocumentNotFoundError, PageIndexError
from ..errors import (AUTH_HINT, CloudAPIError, CollectionNotFoundError,
DocumentNotFoundError, PageIndexError)
from ..events import QueryEvent
logger = logging.getLogger(__name__)
@ -30,8 +31,9 @@ def _as_int(value):
class CloudBackend:
def __init__(self, api_key: str):
def __init__(self, api_key: str, base_url: str | None = None):
self._api_key = api_key
self._base_url = base_url or API_BASE
self._headers = {"api_key": api_key}
self._folder_id_cache: dict[str, str | None] = {}
self._folder_warning_shown = False
@ -59,7 +61,7 @@ class CloudBackend:
"""HTTP helper. ``retries`` caps total attempts — pass 1 for
non-idempotent, expensive calls (e.g. chat completions) where a
retry would redo the full server-side work."""
url = f"{API_BASE}{path}"
url = f"{self._base_url}{path}"
kwargs.setdefault("timeout", 30)
last_status: int | None = None
for attempt in range(retries):
@ -150,7 +152,9 @@ class CloudBackend:
raise
def _get_folder_id(self, name: str) -> str | None:
"""Resolve collection name to folder ID. Returns None if folders not available.
"""Resolve collection name to folder ID. Returns None if folders are
not available on this plan; raises CollectionNotFoundError if folders
are available but no folder has this name.
Only "folders unavailable on this plan" (403/404) is cached as None
transient errors (network, 5xx) propagate so a blip can't silently
@ -170,7 +174,10 @@ class CloudBackend:
if folder.get("name") == name:
self._folder_id_cache[name] = folder["id"]
return folder["id"]
return None
raise CollectionNotFoundError(
f"Collection '{name}' does not exist; "
f"create it first (e.g. client.collection('{name}'))."
)
def list_collections(self) -> list[str]:
try:
@ -183,7 +190,10 @@ class CloudBackend:
return [f["name"] for f in data.get("folders", []) or []]
def delete_collection(self, name: str) -> None:
folder_id = self._get_folder_id(name)
try:
folder_id = self._get_folder_id(name)
except CollectionNotFoundError:
return # already gone — delete is idempotent
if folder_id:
self._request("DELETE", f"/folder/{self._enc(folder_id)}/")
# Drop the cached id so a later same-name op re-resolves instead of
@ -382,10 +392,13 @@ class CloudBackend:
raise ValueError(
"doc_ids cannot be empty; pass None to query the whole collection"
)
doc_id = doc_ids if doc_ids else self._get_all_doc_ids(collection)
doc_id = doc_ids if doc_ids else await asyncio.to_thread(
self._get_all_doc_ids, collection
)
if not doc_id:
raise ValueError("collection has no documents to query")
headers = self._headers
base_url = self._base_url
# Queue carries QueryEvent, an Exception to re-raise, or None (end).
queue: asyncio.Queue[QueryEvent | Exception | None] = asyncio.Queue()
loop = asyncio.get_running_loop()
@ -413,7 +426,7 @@ class CloudBackend:
answer_parts: list[str] = []
try:
resp = requests.post(
f"{API_BASE}/chat/completions/",
f"{base_url}/chat/completions/",
headers=headers,
json={
"messages": [{"role": "user", "content": question}],
@ -521,7 +534,11 @@ class CloudBackend:
# during teardown; closing is just to unblock the thread.
logger.debug("Ignoring error closing streaming response during cleanup",
exc_info=True)
thread.join(timeout=5)
try:
await asyncio.to_thread(thread.join, 5)
except RuntimeError:
# event loop already shutting down — fall back to a bounded sync join
thread.join(timeout=5)
def _get_all_doc_ids(self, collection: str) -> list[str]:
"""Get all document IDs in a collection."""

View file

@ -137,7 +137,7 @@ class LocalBackend:
"structure": result["structure"],
"pages": pages,
})
except sqlite3.IntegrityError:
except sqlite3.IntegrityError as e:
# Lost a concurrent add of the same content (UNIQUE collection+hash).
# Discard our managed files and return the winner's doc_id.
managed_path.unlink(missing_ok=True)
@ -147,7 +147,12 @@ class LocalBackend:
existing_id = self._storage.find_document_by_hash(collection, file_hash)
if existing_id:
return existing_id
raise
# No winner — likely an FK violation from a concurrent collection delete.
if collection not in self._storage.list_collections():
raise CollectionNotFoundError(
f"Collection '{collection}' was deleted while indexing {file_path}"
) from e
raise IndexingError(f"Failed to index {file_path}: {e}") from e
except Exception as e:
managed_path.unlink(missing_ok=True)
doc_dir = col_dir / doc_id

View file

@ -74,7 +74,7 @@ class PageIndexClient:
def _init_cloud(self, api_key: str):
from .backend.cloud import CloudBackend
from .cloud_api import LegacyCloudAPI
self._backend = CloudBackend(api_key=api_key)
self._backend = CloudBackend(api_key=api_key, base_url=self.BASE_URL)
self._legacy_cloud_api = LegacyCloudAPI(api_key=api_key, base_url=self.BASE_URL)
def _init_local(self, model: str = None, retrieve_model: str = None,

View file

@ -166,12 +166,15 @@ class LegacyCloudAPI:
if stream_metadata:
payload["stream_metadata"] = stream_metadata
# Non-streaming completions return no bytes until server-side
# generation finishes — far longer than the default 30s read timeout.
response = self._request(
"POST",
"/chat/completions/",
"Failed to get chat completion",
json=payload,
stream=stream,
**({} if stream else {"timeout": 300}),
)
if stream:

View file

@ -28,7 +28,8 @@ class IndexConfig(BaseModel):
if_add_node_text: bool = False
# Max concurrent in-flight LLM calls during indexing. None = use the global
# default (get_max_concurrency(), overridable via PAGEINDEX_MAX_CONCURRENCY).
# An explicit value here wins for this client.
# An explicit value can only lower the cap; raise the ceiling itself via
# set_max_concurrency() or PAGEINDEX_MAX_CONCURRENCY.
max_concurrency: int | None = None
# Per-call litellm completion kwargs for this client's indexing calls only
# (e.g. {"temperature": 1}). None = use the process-wide defaults
@ -217,6 +218,16 @@ def max_concurrency_scope(value: int | None):
"""
if value is not None:
_validate_max_concurrency(value)
if value > _MAX_CONCURRENCY:
import warnings
warnings.warn(
f"max_concurrency={value} exceeds the process-wide ceiling "
f"({_MAX_CONCURRENCY}), which still applies — a per-index value "
f"can only lower the cap. Raise the ceiling with "
f"set_max_concurrency({value}) or PAGEINDEX_MAX_CONCURRENCY.",
UserWarning,
stacklevel=3,
)
scoped_sem = threading.Semaphore(value) if value is not None else None
token = _MAX_CONCURRENCY_OVERRIDE.set(value)
sem_token = _MAX_CONCURRENCY_SCOPE_SEMAPHORE.set(scoped_sem)

View file

@ -17,6 +17,9 @@ async def check_title_appearance(item, page_list, start_index=1, model=None):
page_number = item['physical_index']
if page_number < start_index:
# a below-range index would wrap to a negative Python index (the last page)
return {'list_index': item.get('list_index'), 'answer': 'no', 'title': title, 'page_number': None}
page_text = page_list[page_number-start_index][0]
@ -1085,6 +1088,13 @@ async def tree_parser(page_list, opt, doc=None, logger=None):
def page_index_main(doc, opt=None):
# accept legacy 'yes'/'no' string flags (a bare 'no' is truthy)
from .page_index_md import _coerce_bool
for flag in ('if_add_node_id', 'if_add_node_text',
'if_add_node_summary', 'if_add_doc_description'):
if hasattr(opt, flag):
setattr(opt, flag, _coerce_bool(getattr(opt, flag)))
logger = JsonLogger(doc)
is_valid_pdf = (

View file

@ -12,3 +12,6 @@ warnings.warn(
)
from .index.utils import * # noqa: F401,F403,E402
# Legacy 0.2.x alias. index.utils keeps it private (_config) so its star-export
# can't shadow the pageindex.config submodule; re-expose it only in this shim.
from types import SimpleNamespace as config # noqa: E402,F401

View file

@ -7,7 +7,7 @@ import pytest
import pageindex.backend.cloud as cloud_mod
from pageindex.backend.cloud import CloudBackend, API_BASE
from pageindex.errors import CloudAPIError, DocumentNotFoundError
from pageindex.errors import CloudAPIError, CollectionNotFoundError, DocumentNotFoundError
# Real sleep captured at import, before the _no_sleep autouse fixture patches
# time.sleep on the shared module object. Tests that need genuine pacing (e.g.
@ -100,6 +100,47 @@ def test_list_documents_paginates(monkeypatch):
assert docs[-1]["doc_id"] == "d129"
# ── base_url override must reach every request path ──────────────────────────
def test_base_url_override_routes_requests(monkeypatch):
backend = CloudBackend(api_key="pi-test", base_url="https://staging.example.com")
urls = []
def fake_request(method, url, headers=None, **kwargs):
urls.append(url)
return FakeResponse(status_code=200, json_data={"folders": []})
monkeypatch.setattr(cloud_mod.requests, "request", fake_request)
backend.list_collections()
assert urls == ["https://staging.example.com/folders/"]
def test_base_url_override_routes_streaming(monkeypatch):
backend = CloudBackend(api_key="pi-test", base_url="https://staging.example.com")
urls = []
def fake_post(url, **kwargs):
urls.append(url)
return FakeResponse(status_code=200, lines=["data: [DONE]"])
monkeypatch.setattr(cloud_mod.requests, "post", fake_post)
_collect_events(backend)
assert urls == ["https://staging.example.com/chat/completions/"]
def test_client_base_url_override_reaches_both_backends():
# PageIndexClient.BASE_URL is the documented override point; it must apply
# to the Collection API (CloudBackend), not just the legacy SDK methods.
from pageindex.client import PageIndexClient
class StagingClient(PageIndexClient):
BASE_URL = "https://staging.example.com"
client = StagingClient(api_key="pi-test")
assert client._backend._base_url == "https://staging.example.com"
assert client._legacy_cloud_api.base_url == "https://staging.example.com"
# ── folder resolution: plan-limit vs transient errors ────────────────────────
def test_folder_unavailable_warns_and_caches(monkeypatch):
@ -127,6 +168,42 @@ def test_folder_transient_error_propagates_and_is_not_cached(monkeypatch):
assert "col" not in backend._folder_id_cache
def test_missing_folder_raises_collection_not_found(monkeypatch):
# Folders ARE available but the name has no match (e.g. deleted): must
# raise, never fall back to the account-wide global space.
backend = CloudBackend(api_key="pi-test")
monkeypatch.setattr(
backend, "_request",
lambda *a, **k: {"folders": [{"name": "other", "id": "f1"}]},
)
with pytest.raises(CollectionNotFoundError):
backend._get_folder_id("gone")
assert "gone" not in backend._folder_id_cache
def test_list_documents_raises_for_missing_collection(monkeypatch):
# Guards the query path: a stale collection must error out, not silently
# list (and query over) every document in the account.
backend = CloudBackend(api_key="pi-test")
monkeypatch.setattr(backend, "_request", lambda *a, **k: {"folders": []})
with pytest.raises(CollectionNotFoundError):
backend.list_documents("gone")
def test_delete_collection_missing_is_noop(monkeypatch):
backend = CloudBackend(api_key="pi-test")
calls = []
def fake_request(method, path, **kwargs):
calls.append((method, path))
return {"folders": []}
monkeypatch.setattr(backend, "_request", fake_request)
backend.delete_collection("gone") # idempotent, matching the local backend
assert ("GET", "/folders/") in calls
assert not any(method == "DELETE" for method, _ in calls)
def test_list_collections_degrades_when_folders_unavailable(monkeypatch):
backend = CloudBackend(api_key="pi-test")

View file

@ -0,0 +1,215 @@
"""Regression tests for the PR #272 review findings #9-#15."""
import asyncio
import sqlite3
import threading
import warnings
import pytest
from pageindex.errors import CollectionNotFoundError, IndexingError
# ── #9: below-range physical_index must not wrap to the last page ────────────
def test_check_title_appearance_rejects_below_range_index(monkeypatch):
from pageindex.index import page_index as pi
async def _fail(*a, **k):
raise AssertionError("LLM must not be called for a below-range index")
monkeypatch.setattr(pi, "llm_acompletion", _fail)
item = {"title": "Intro", "physical_index": 0, "list_index": 3}
# A single-page page_list: without the guard, page_list[0-1] silently
# reads this (last) page instead of erroring.
result = asyncio.run(
pi.check_title_appearance(item, [("last page text", 10)], start_index=1)
)
assert result["answer"] == "no"
assert result["page_number"] is None
# ── #6: page_index_main must coerce legacy 'yes'/'no' string flags ───────────
def test_page_index_main_coerces_legacy_string_flags():
from types import SimpleNamespace
from pageindex.index.page_index import page_index_main
opt = SimpleNamespace(model=None, if_add_node_id='yes', if_add_node_text='no',
if_add_node_summary='no', if_add_doc_description='no')
# Coercion runs before input validation, which rejects the non-PDF path.
with pytest.raises(ValueError, match="Unsupported input type"):
page_index_main('not-a-pdf.txt', opt)
assert opt.if_add_node_id is True
assert opt.if_add_node_text is False
assert opt.if_add_node_summary is False
assert opt.if_add_doc_description is False
# ── #10: per-index max_concurrency above the ceiling must warn ────────────────
def test_max_concurrency_scope_warns_above_ceiling():
from pageindex.config import (
max_concurrency_scope,
set_max_concurrency,
_process_wide_max_concurrency,
)
original = _process_wide_max_concurrency()
set_max_concurrency(5)
try:
with pytest.warns(UserWarning, match="exceeds the process-wide ceiling"):
with max_concurrency_scope(20):
pass
# A narrowing value is the supported use and must stay silent.
with warnings.catch_warnings():
warnings.simplefilter("error")
with max_concurrency_scope(3):
pass
finally:
set_max_concurrency(original)
# ── #11: non-streaming legacy chat_completions needs a long read timeout ─────
def test_legacy_chat_completions_nonstream_timeout(monkeypatch):
from pageindex import cloud_api as ca
captured = {}
class FakeResp:
status_code = 200
text = ""
def json(self):
return {"choices": []}
def fake_request(method, url, headers=None, **kwargs):
captured.clear()
captured.update(kwargs)
return FakeResp()
monkeypatch.setattr(ca.requests, "request", fake_request)
api = ca.LegacyCloudAPI(api_key="pi-test")
api.chat_completions(messages=[{"role": "user", "content": "q"}], stream=False)
assert captured["timeout"] == 300
api.chat_completions(messages=[{"role": "user", "content": "q"}], stream=True)
assert captured["timeout"] == 120 # between-chunks timeout, unchanged
# ── #12: query_stream must not run the doc-id listing on the loop thread ─────
def test_query_stream_lists_docs_off_event_loop(monkeypatch):
import pageindex.backend.cloud as cloud_mod
from pageindex.backend.cloud import CloudBackend
backend = CloudBackend(api_key="pi-test")
seen = {}
def fake_get_all(collection):
seen["thread"] = threading.current_thread()
return ["d1"]
monkeypatch.setattr(backend, "_get_all_doc_ids", fake_get_all)
class FakeResponse:
status_code = 200
text = ""
def iter_lines(self, decode_unicode=True):
yield "data: [DONE]"
def close(self):
pass
monkeypatch.setattr(cloud_mod.requests, "post", lambda *a, **k: FakeResponse())
async def _run():
async for _ in backend.query_stream("col", "q"): # doc_ids=None → list all
pass
return threading.current_thread()
loop_thread = asyncio.run(_run())
assert seen["thread"] is not loop_thread
# ── #13: an unexplained IntegrityError must surface as a PageIndexError ──────
class _RaceStorage:
"""Collection exists at the fail-fast check, then save trips the FK."""
def __init__(self, collections_after_start):
self._after = collections_after_start
self._calls = 0
def list_collections(self):
self._calls += 1
return ["col"] if self._calls == 1 else self._after
def find_document_by_hash(self, collection, file_hash):
return None
def save_document(self, *a, **k):
raise sqlite3.IntegrityError("FOREIGN KEY constraint failed")
def _make_backend(tmp_path, monkeypatch, storage):
import pageindex.backend.local as local_mod
from pageindex.backend.local import LocalBackend
backend = LocalBackend(storage=storage, files_dir=str(tmp_path / "files"))
class FakeParsed:
doc_name = "doc"
nodes = []
class FakeParser:
def parse(self, path, model=None, images_dir=None):
return FakeParsed()
monkeypatch.setattr(backend, "_resolve_parser", lambda p: FakeParser())
monkeypatch.setattr(
local_mod, "build_index", lambda parsed, model=None, opt=None: {"structure": []}
)
pdf = tmp_path / "doc.pdf"
pdf.write_bytes(b"%PDF-1.4 fake")
return backend, str(pdf)
def test_concurrent_collection_delete_raises_collection_not_found(tmp_path, monkeypatch):
backend, pdf = _make_backend(tmp_path, monkeypatch, _RaceStorage([]))
with pytest.raises(CollectionNotFoundError):
backend.add_document("col", pdf)
def test_unexplained_integrity_error_wrapped_as_indexing_error(tmp_path, monkeypatch):
backend, pdf = _make_backend(tmp_path, monkeypatch, _RaceStorage(["col"]))
with pytest.raises(IndexingError):
backend.add_document("col", pdf)
# ── #14: legacy `from pageindex.utils import config` must keep working ───────
def test_legacy_config_alias_importable():
from types import SimpleNamespace
with warnings.catch_warnings():
warnings.simplefilter("ignore") # module-level deprecation shim warning
from pageindex.utils import config
assert config is SimpleNamespace
# ── #15: star import must bind the legacy pre-SDK names ──────────────────────
def test_star_import_binds_legacy_names():
ns = {}
exec("from pageindex import *", ns)
for name in (
"page_index",
"page_index_main",
"tree_parser",
"ConfigLoader",
"llm_completion",
"llm_acompletion",
):
assert name in ns, f"{name} missing from star import"