fix(ocr): wire batch settings into _field_map + review nits
Round 1 review (PR #910): - BLOCKING: add document_ocr_mode / _batch_poll_seconds / _batch_max_wait_seconds to config._field_map — without it dynaconf silently ignored the env vars and DOCUMENT_OCR_MODE=batch could never be enabled in production. Add a regression test asserting the three round-trip from env. - migration 008: give batch_ocr_jobs a composite PRIMARY KEY on (user_id, doc_id, doc_type, etag) instead of a bare UniqueConstraint (N1). - OcrProcessor: use a dedicated _batch_client_lock instead of sharing the sync backend lock (N3). - tests: use https:// gateway URLs in the new fixtures to clear SonarCloud's "insecure http" security hotspots (all 14 were test-only http://gw literals). 1653 unit tests pass; ruff + ty green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
3b7e8d779b
commit
2b7dfc8535
@@ -36,7 +36,11 @@ def upgrade() -> None:
|
||||
# Document identity (the same keys the OCR tier receives via the
|
||||
# processor ``options``). ``etag`` is the content-version key: a changed
|
||||
# document (new etag) is a new job, so a stale row never serves results
|
||||
# for the wrong content.
|
||||
# for the wrong content. The four-column natural key IS the primary key:
|
||||
# one in-flight job per (document, content version), and the PK doubles as
|
||||
# the unique index ``insert_pending``'s ON CONFLICT target relies on. A
|
||||
# resubmit for a new etag inserts a new row; the superseded row is swept on
|
||||
# resubmit (delete_stale_for_doc).
|
||||
sa.Column("user_id", sa.Text(), nullable=False),
|
||||
sa.Column("doc_id", sa.Text(), nullable=False),
|
||||
sa.Column("doc_type", sa.Text(), nullable=False),
|
||||
@@ -51,14 +55,8 @@ def upgrade() -> None:
|
||||
# (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS).
|
||||
sa.Column("submitted_at", sa.BigInteger(), nullable=False),
|
||||
sa.Column("updated_at", sa.BigInteger(), nullable=False),
|
||||
# One in-flight job per (document, content version). A resubmit for a new
|
||||
# etag inserts a new row; the superseded row is swept on resubmit.
|
||||
sa.UniqueConstraint(
|
||||
"user_id",
|
||||
"doc_id",
|
||||
"doc_type",
|
||||
"etag",
|
||||
name="uq_batch_ocr_jobs_doc",
|
||||
sa.PrimaryKeyConstraint(
|
||||
"user_id", "doc_id", "doc_type", "etag", name="pk_batch_ocr_jobs"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -1521,6 +1521,9 @@ def get_settings() -> Settings:
|
||||
"document_ocr_provider": "DOCUMENT_OCR_PROVIDER",
|
||||
"document_ocr_model": "DOCUMENT_OCR_MODEL",
|
||||
"document_ocr_timeout_seconds": "DOCUMENT_OCR_TIMEOUT_SECONDS",
|
||||
"document_ocr_mode": "DOCUMENT_OCR_MODE",
|
||||
"document_ocr_batch_poll_seconds": "DOCUMENT_OCR_BATCH_POLL_SECONDS",
|
||||
"document_ocr_batch_max_wait_seconds": "DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS",
|
||||
"document_ocr_min_text_quality": "DOCUMENT_OCR_MIN_TEXT_QUALITY",
|
||||
"document_ocr_page_fraction": "DOCUMENT_OCR_PAGE_FRACTION",
|
||||
"document_ocr_min_page_chars": "DOCUMENT_OCR_MIN_PAGE_CHARS",
|
||||
|
||||
@@ -272,6 +272,7 @@ class OcrProcessor(DocumentProcessor):
|
||||
# using sync" warning to once per pod.
|
||||
self._batch_client_resolved = False
|
||||
self._batch_client: Any = None
|
||||
self._batch_client_lock: anyio.Lock | None = None
|
||||
self._batch_fallback_warned = False
|
||||
|
||||
@property
|
||||
@@ -377,9 +378,9 @@ class OcrProcessor(DocumentProcessor):
|
||||
provider=mistral / no gateway). Resolved once under the backend lock so the
|
||||
token provider's M2M cache survives across documents."""
|
||||
if not self._batch_client_resolved:
|
||||
if self._backend_lock is None:
|
||||
self._backend_lock = anyio.Lock()
|
||||
async with self._backend_lock:
|
||||
if self._batch_client_lock is None:
|
||||
self._batch_client_lock = anyio.Lock()
|
||||
async with self._batch_client_lock:
|
||||
if not self._batch_client_resolved: # double-checked
|
||||
self._batch_client = build_gateway_batch_client(get_settings())
|
||||
self._batch_client_resolved = True
|
||||
|
||||
@@ -115,6 +115,28 @@ class TestGetSettings:
|
||||
assert settings.oidc_token_type == "jwt"
|
||||
assert settings.oidc_scopes == "openid profile"
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"DOCUMENT_OCR_MODE": "batch",
|
||||
"DOCUMENT_OCR_BATCH_POLL_SECONDS": "45",
|
||||
"DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS": "3600",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
def test_get_settings_ocr_batch_mode_from_env(self):
|
||||
"""DOCUMENT_OCR_MODE / batch tuning must reach settings (regression).
|
||||
|
||||
These were added to _DEFAULTS + the Settings dataclass but initially
|
||||
omitted from _field_map, so dynaconf silently ignored the env vars and
|
||||
batch mode could never be enabled in production (Deck #332).
|
||||
"""
|
||||
_reload_config()
|
||||
settings = get_settings()
|
||||
assert settings.document_ocr_mode == "batch"
|
||||
assert settings.document_ocr_batch_poll_seconds == 45
|
||||
assert settings.document_ocr_batch_max_wait_seconds == 3600
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{"QDRANT_LOCATION": "/app/data/qdrant"},
|
||||
|
||||
@@ -33,8 +33,8 @@ def _patch_transport(monkeypatch, handler) -> list[httpx.Request]:
|
||||
|
||||
|
||||
def test_base_url_normalization():
|
||||
assert gbc.GatewayBatchOcrClient("http://gw", "m")._base == "http://gw/v1"
|
||||
assert gbc.GatewayBatchOcrClient("http://gw/v1/", "m")._base == "http://gw/v1"
|
||||
assert gbc.GatewayBatchOcrClient("https://gw", "m")._base == "https://gw/v1"
|
||||
assert gbc.GatewayBatchOcrClient("https://gw/v1/", "m")._base == "https://gw/v1"
|
||||
|
||||
|
||||
async def test_submit_posts_one_document_and_returns_job_id(monkeypatch):
|
||||
@@ -44,7 +44,7 @@ async def test_submit_posts_one_document_and_returns_job_id(monkeypatch):
|
||||
)
|
||||
|
||||
seen = _patch_transport(monkeypatch, handler)
|
||||
client = gbc.GatewayBatchOcrClient("http://gw", "mistral/mistral-ocr-latest")
|
||||
client = gbc.GatewayBatchOcrClient("https://gw", "mistral/mistral-ocr-latest")
|
||||
|
||||
job_id = await client.submit(b"%PDF-1.7", "application/pdf", custom_id="doc-9")
|
||||
|
||||
@@ -73,7 +73,7 @@ async def test_submit_sends_bearer_when_token_provider(monkeypatch):
|
||||
# _Tok duck-types get_token; cast for the type checker (the client only awaits
|
||||
# get_token()).
|
||||
client = gbc.GatewayBatchOcrClient(
|
||||
"http://gw", "m", token_provider=cast(Any, _Tok())
|
||||
"https://gw", "m", token_provider=cast(Any, _Tok())
|
||||
)
|
||||
await client.submit(b"x", "application/pdf", custom_id="d")
|
||||
assert seen[0].headers["Authorization"] == "Bearer tok-abc"
|
||||
@@ -84,7 +84,7 @@ async def test_poll_pending(monkeypatch):
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "pending", "total": 1}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_pending and result.pages == []
|
||||
|
||||
|
||||
@@ -102,7 +102,7 @@ async def test_poll_succeeded_maps_pages(monkeypatch):
|
||||
],
|
||||
}
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body))
|
||||
result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_succeeded
|
||||
# Order is preserved as returned; _pages_to_text sorts downstream.
|
||||
assert result.pages == [(1, "two"), (0, "one")]
|
||||
@@ -113,14 +113,14 @@ async def test_poll_failed_surfaces_error(monkeypatch):
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "failed", "error": "quota"}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed and result.error == "quota"
|
||||
|
||||
|
||||
async def test_poll_succeeded_with_per_document_error_is_failed(monkeypatch):
|
||||
body = {"status": "succeeded", "results": [{"custom_id": "d", "error": "bad page"}]}
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body))
|
||||
result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed and result.error == "bad page"
|
||||
|
||||
|
||||
@@ -129,7 +129,7 @@ async def test_poll_succeeded_no_results_is_failed(monkeypatch):
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "succeeded", "results": []}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed
|
||||
|
||||
|
||||
@@ -138,4 +138,4 @@ async def test_poll_raises_on_http_error(monkeypatch):
|
||||
monkeypatch, lambda r: httpx.Response(503, json={"detail": "down"})
|
||||
)
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j")
|
||||
await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
|
||||
@@ -281,7 +281,7 @@ def _wire_batch(monkeypatch, *, client, store, settings=None):
|
||||
settings = settings or _settings(
|
||||
document_ocr_mode="batch",
|
||||
document_ocr_provider="gateway",
|
||||
embedding_gateway_url="http://gw",
|
||||
embedding_gateway_url="https://gw",
|
||||
)
|
||||
monkeypatch.setattr(ocr, "get_settings", lambda: settings)
|
||||
monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client)
|
||||
@@ -404,7 +404,7 @@ async def test_batch_falls_back_to_sync_when_no_identity(monkeypatch):
|
||||
settings = _settings(
|
||||
document_ocr_mode="batch",
|
||||
document_ocr_provider="gateway",
|
||||
embedding_gateway_url="http://gw",
|
||||
embedding_gateway_url="https://gw",
|
||||
)
|
||||
monkeypatch.setattr(ocr, "get_settings", lambda: settings)
|
||||
monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client)
|
||||
|
||||
Reference in New Issue
Block a user