refactor: extract the per-source release search out of /api/releases (#1355)

The `/api/releases` route carries an inner `_search_source_releases`
helper that builds the search plan for one source, logs the planned
query type, runs the search and turns
`SourceUnavailableError`/operational errors into an error message
instead of raising. Anything outside the route that wants to search one
source with exactly those semantics has to go through Flask today.

This moves that helper into `shelfmark/core/release_search.py` as
`search_source_releases()` and has the route delegate to it. Behaviour
is unchanged: same plan construction (including the caller's `user_id`,
so per-user default languages still apply), same logging, same
error-to-message handling.

It is the refactor half of #1047 by @InfiniteAvenger, split out on its
own as you asked for other PRs (#1318). Their authorship is preserved on
the commit; I rebased it onto current `main` and added tests.

## Verification

- `tests/core/test_release_search.py`: unknown source → `"Unknown
source: …"`, `SourceUnavailableError` and operational errors →
`"<source>: <error>"`, success path forwards `expand_search` /
`content_type` and returns the source instance, the plan receives
languages / manual query / indexers / `user_id`. (These tests are
type-annotated; happy to strip the annotations if you prefer the suite's
bare style.)
- Full suite, ruff, ruff format, basedpyright, vulture green; the
existing `/api/releases` route tests are unchanged and pass.

Co-authored-by: InfiniteAvenger <calebewest02@gmail.com>

shelfmark/core/release_search.py

@@ -0,0 +1,93 @@
+"""Shared release-search helpers.
+
+Extracted from the ``/api/releases`` route so the same per-source search logic can
+be reused outside the HTTP route (for example by background automation) without
+going through Flask. Behaviour for the HTTP route is preserved: the route delegates
+its inner per-source search to :func:`search_source_releases`.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING
+
+from shelfmark.core.logger import setup_logger
+from shelfmark.core.search_plan import build_release_search_plan
+
+if TYPE_CHECKING:
+    from shelfmark.core.models import SearchFilters
+    from shelfmark.metadata_providers import BookMetadata
+    from shelfmark.release_sources import Release, ReleaseSource
+
+logger = setup_logger(__name__)
+
+# Mirror of main._OPERATIONAL_ERRORS so a misbehaving source can't crash a caller.
+_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
+
+
+def search_source_releases(
+    source_name: str,
+    search_book: BookMetadata,
+    *,
+    languages: list[str] | None = None,
+    manual_query: str | None = None,
+    indexers: list[str] | None = None,
+    expand_search: bool = False,
+    content_type: str = "ebook",
+    source_filters: SearchFilters | None = None,
+    user_id: int | None = None,
+) -> tuple[ReleaseSource | None, list[Release], str | None]:
+    """Search a single release source, returning any error instead of raising.
+
+    Returns ``(source, releases, error_message)``. On failure ``source`` is ``None``
+    and ``error_message`` describes the problem. ``user_id`` lets the search plan
+    pick up that user's default languages when no explicit filter is given.
+    """
+    from shelfmark.release_sources import SourceUnavailableError, get_source
+
+    try:
+        source = get_source(source_name)
+
+        plan = build_release_search_plan(
+            search_book,
+            languages=languages,
+            manual_query=manual_query,
+            indexers=indexers,
+            source_filters=source_filters,
+            user_id=user_id,
+        )
+
+        if plan.source_filters is not None:
+            planned_query = plan.manual_query or plan.primary_query
+            planned_query_type = "query"
+        elif plan.manual_query:
+            planned_query = plan.manual_query
+            planned_query_type = "manual"
+        elif not expand_search and plan.isbn_candidates:
+            planned_query = plan.isbn_candidates[0]
+            planned_query_type = "isbn"
+        else:
+            planned_query = plan.primary_query
+            planned_query_type = "title_author"
+
+        logger.debug(
+            "Searching %s: %s='%s' (title='%s', authors=%s, expand=%s, content_type=%s)",
+            source_name,
+            planned_query_type,
+            planned_query,
+            search_book.title,
+            search_book.authors,
+            expand_search,
+            content_type,
+        )
+
+        releases = source.search(
+            search_book, plan, expand_search=expand_search, content_type=content_type
+        )
+    except ValueError:
+        return None, [], f"Unknown source: {source_name}"
+    except (SourceUnavailableError, *_OPERATIONAL_ERRORS) as exc:
+        logger.warning("Release search failed for source %s: %s", source_name, exc)
+        return None, [], f"{source_name}: {exc!s}"
+    else:
+        return source, releases, None

shelfmark/main.py

@@ -2829,7 +2829,7 @@ def api_releases() -> Response | tuple[Response, int]:
     try:
         from dataclasses import asdict
 
-        from shelfmark.core.search_plan import build_release_search_plan
+        from shelfmark.core.release_search import search_source_releases
         from shelfmark.metadata_providers import (
             BookMetadata,
             get_provider,
@@ -2848,54 +2848,17 @@ def api_releases() -> Response | tuple[Response, int]:
             source_name: str, search_book: BookMetadata
         ) -> tuple[Any | None, list[Any], str | None]:
             """Search one source and return any error message instead of raising."""
-            try:
-                source = get_source(source_name)
-
-                plan = build_release_search_plan(
-                    search_book,
-                    languages=browse_filters.lang
-                    if source_query_filters is not None
-                    else languages,
-                    manual_query=query_text if source_query_filters is not None else manual_query,
-                    indexers=indexers,
-                    source_filters=source_query_filters,
-                    user_id=db_user_id,
-                )
-
-                if plan.source_filters is not None:
-                    planned_query = plan.manual_query or plan.primary_query
-                    planned_query_type = "query"
-                elif plan.manual_query:
-                    planned_query = plan.manual_query
-                    planned_query_type = "manual"
-                elif not expand_search and plan.isbn_candidates:
-                    planned_query = plan.isbn_candidates[0]
-                    planned_query_type = "isbn"
-                else:
-                    planned_query = plan.primary_query
-                    planned_query_type = "title_author"
-
-                logger.debug(
-                    "Searching %s: %s='%s' (title='%s', authors=%s, expand=%s, content_type=%s)",
-                    source_name,
-                    planned_query_type,
-                    planned_query,
-                    search_book.title,
-                    search_book.authors,
-                    expand_search,
-                    content_type,
-                )
-
-                releases = source.search(
-                    search_book, plan, expand_search=expand_search, content_type=content_type
-                )
-            except ValueError:
-                return None, [], f"Unknown source: {source_name}"
-            except (SourceUnavailableError, *_OPERATIONAL_ERRORS) as e:
-                logger.warning("Release search failed for source %s: %s", source_name, e)
-                return None, [], f"{source_name}: {e!s}"
-            else:
-                return source, releases, None
+            return search_source_releases(
+                source_name,
+                search_book,
+                languages=(browse_filters.lang if source_query_filters is not None else languages),
+                manual_query=(query_text if source_query_filters is not None else manual_query),
+                indexers=indexers,
+                expand_search=expand_search,
+                content_type=content_type,
+                source_filters=source_query_filters,
+                user_id=db_user_id,
+            )
 
         provider = request.args.get("provider", "").strip()
         book_id = request.args.get("book_id", "").strip()

tests/core/test_release_search.py

@@ -0,0 +1,119 @@
+"""Tests for the per-source release search helper extracted from ``/api/releases``."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from shelfmark.core import release_search
+from shelfmark.metadata_providers import BookMetadata
+from shelfmark.release_sources import SourceUnavailableError
+
+
+def _book(**overrides: Any) -> BookMetadata:
+    fields: dict[str, Any] = {
+        "provider": "manual",
+        "provider_id": "1",
+        "title": "Dungeon Crawler Carl",
+        "authors": ["Matt Dinniman"],
+        "isbn_13": "9780593820247",
+    }
+    fields.update(overrides)
+    return BookMetadata(**fields)
+
+
+class _Source:
+    def __init__(self, result: list[str] | None = None, error: Exception | None = None) -> None:
+        self.result = result if result is not None else []
+        self.error = error
+        self.calls: list[tuple[Any, Any, bool, str]] = []
+
+    def search(
+        self, book: BookMetadata, plan: Any, *, expand_search: bool, content_type: str
+    ) -> list[str]:
+        self.calls.append((book, plan, expand_search, content_type))
+        if self.error is not None:
+            raise self.error
+        return list(self.result)
+
+
+def test_unknown_source_returns_error_tuple(monkeypatch: pytest.MonkeyPatch) -> None:
+    def _raise(name: str) -> Any:
+        msg = f"Unknown source: {name}"
+        raise ValueError(msg)
+
+    monkeypatch.setattr("shelfmark.release_sources.get_source", _raise)
+
+    source, releases, error = release_search.search_source_releases("nope", _book())
+
+    assert source is None
+    assert releases == []
+    assert error == "Unknown source: nope"
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [SourceUnavailableError("down"), OSError("boom"), RuntimeError("bad")],
+    ids=["unavailable", "oserror", "runtime"],
+)
+def test_source_failures_become_messages(monkeypatch: pytest.MonkeyPatch, exc: Exception) -> None:
+    src = _Source(error=exc)
+    monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src)
+
+    source, releases, error = release_search.search_source_releases("s", _book())
+
+    assert source is None
+    assert releases == []
+    assert error == f"s: {exc!s}"
+
+
+def test_success_returns_source_and_forwards_search_kwargs(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    src = _Source(result=["r1", "r2"])
+    monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src)
+
+    source, releases, error = release_search.search_source_releases(
+        "s", _book(), expand_search=True, content_type="audiobook"
+    )
+
+    assert source is src
+    assert releases == ["r1", "r2"]
+    assert error is None
+    book, plan, expand_search, content_type = src.calls[0]
+    assert book.title == "Dungeon Crawler Carl"
+    assert "dungeon crawler carl" in plan.primary_query.lower()
+    assert plan.isbn_candidates == ["9780593820247"]
+    assert expand_search is True
+    assert content_type == "audiobook"
+
+
+def test_plan_receives_filters_and_user_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    real_build = release_search.build_release_search_plan
+
+    def _build(book: BookMetadata, **kwargs: Any) -> Any:
+        captured.update(kwargs)
+        return real_build(book, **kwargs)
+
+    monkeypatch.setattr(release_search, "build_release_search_plan", _build)
+    src = _Source()
+    monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src)
+
+    release_search.search_source_releases(
+        "s",
+        _book(),
+        languages=["en"],
+        manual_query="dcc",
+        indexers=["idx"],
+        user_id=42,
+    )
+
+    assert captured == {
+        "languages": ["en"],
+        "manual_query": "dcc",
+        "indexers": ["idx"],
+        "source_filters": None,
+        "user_id": 42,
+    }