refactor: move the metadata routes into core/metadata_routes.py (#1623)

The metadata routes are the next group from the plan in Discussion
#1536: the eight `/api/metadata/...` handlers (providers, config,
search, field-options, book and the three targets routes) move from
`main.py` to `core/metadata_routes.py`, registered with
`register_metadata_routes(app, ...)`. `main.py` is about 490 lines
shorter.

Same two-commit layout as the settings move (#1595). The first commit
adds 39 tests that pass on current `main` before anything moves. They
cover the 401 on all eight routes, the shape of the providers list and
the config, the search errors, the book lookup (200, 404, 400 and 503),
and how target errors map to 400, 502 and 500. The second commit is the
move, and no test changed in it.

Three helpers moved with the routes because nothing else uses them.
`_attach_library_ownership` stays in `main.py` since the releases routes
need it too, and is passed in along with the login decorator and the
auth mode, the way `activity_routes` takes them. The auth mode is a
callable evaluated per request, so the one test that patches it on
`main` still works.

The Flask URL map (every rule, its methods and its endpoint name) is
identical before and after, all 77 rules. The full suite, ruff and
basedpyright are clean. I also ran the app from this branch and hit the
routes over HTTP: providers and config answer 200, a book that doesn't
exist gets a 404, and search answers 503 with the "no metadata provider
configured" message on an instance with none enabled. I did not look at
it in a browser or run the e2e tests.

Log lines for these routes now come from
`shelfmark.core.metadata_routes`, and the error tuples are local copies
so the module doesn't import `main`. One thing to know when reading the
diff: the register call sits just after `_attach_library_ownership` and
not where the first route used to be, because that function has to be
defined before it can be passed in.
This commit is contained in:
splitsec2 authored and GitHub committed 2026-11-01 14:11:33 -07:00
1 parent 0a7e114681
commit d41fa98749
3 files changed
+897 -500

No files matched your search

+541
View File
@@ -0,0 +1,541 @@
"""Metadata providers: the ``/api/metadata/...`` search, book and target routes."""
from __future__ import annotations
import sqlite3
from contextlib import suppress
from functools import wraps
from typing import TYPE_CHECKING, Any
from flask import jsonify, request, session
from shelfmark.core.config import config as app_config
from shelfmark.core.logger import setup_logger
from shelfmark.core.request_helpers import get_session_db_user_id
if TYPE_CHECKING:
from collections.abc import Callable
from flask import Flask
from werkzeug.wrappers import Response
from shelfmark.metadata_providers import BookMetadata, MetadataProvider
logger = setup_logger(__name__)
_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
_IMPORT_OPERATIONAL_ERRORS = (ImportError, *_OPERATIONAL_ERRORS)
def _resolve_metadata_provider(provider_name: str) -> MetadataProvider:
"""Validate, instantiate and return a ready metadata provider.
Raises appropriate HTTP-friendly exceptions on failure.
"""
from shelfmark.metadata_providers import (
get_provider,
get_provider_kwargs,
is_provider_registered,
)
if not is_provider_registered(provider_name):
msg = f"Unknown metadata provider: {provider_name}"
raise ValueError(msg)
kwargs = get_provider_kwargs(provider_name)
prov = get_provider(provider_name, **kwargs)
if not prov.is_available():
msg = f"Provider '{provider_name}' is not available"
raise RuntimeError(msg)
return prov
def _handle_target_errors(
fallback_message: str,
) -> Callable[
[Callable[..., Response | tuple[Response, int]]], Callable[..., Response | tuple[Response, int]]
]:
"""Wrap a metadata-target route with standard error handling."""
def decorator(
fn: Callable[..., Response | tuple[Response, int]],
) -> Callable[..., Response | tuple[Response, int]]:
@wraps(fn)
def wrapper(*args: object, **kwargs: object) -> Response | tuple[Response, int]:
try:
return fn(*args, **kwargs)
except (NotImplementedError, ValueError) as e:
return jsonify({"error": str(e)}), 400
except RuntimeError as e:
return jsonify({"error": str(e)}), 502
except (OSError, TypeError, sqlite3.Error) as e:
logger.error_trace(f"{fallback_message}: {e}")
return jsonify({"error": fallback_message}), 500
return wrapper
return decorator
def register_metadata_routes(
app: Flask,
*,
login_required: Callable[..., Callable[..., Response | tuple[Response, int]]],
resolve_auth_mode: Callable[[], str],
attach_library_ownership: Callable[[BookMetadata, dict[str, Any]], None],
) -> None:
"""Register the ``/api/metadata/...`` routes.
``login_required`` is the app's decorator. ``resolve_auth_mode`` is called per request so a
runtime patch of the auth mode still takes effect. ``attach_library_ownership`` is shared
with the releases routes, which is why it stays in ``main.py``.
"""
def _session_may_use_provider_account() -> bool:
"""Whether this session may read or change the metadata provider's own account.
A provider such as Hardcover is signed in as one account, the instance owner's, and its lists
and shelves belong to that person. With no authentication configured the one user is the owner.
Otherwise only an admin may use them.
"""
return resolve_auth_mode() == "none" or bool(session.get("is_admin", False))
@app.route("/api/metadata/providers", methods=["GET"])
@login_required
def api_metadata_providers() -> Response | tuple[Response, int]:
"""Get list of available metadata providers.
Returns:
flask.Response: JSON with list of providers and their status.
"""
try:
from shelfmark.metadata_providers import (
get_configured_provider_name,
get_provider,
get_provider_kwargs,
list_providers,
)
app_config.refresh()
db_user_id = get_session_db_user_id(session)
configured_metadata_provider = get_configured_provider_name(
content_type="ebook",
user_id=db_user_id,
fallback_to_main=True,
)
configured_audiobook_metadata_provider = get_configured_provider_name(
content_type="audiobook",
user_id=db_user_id,
fallback_to_main=False,
)
configured_combined_metadata_provider = get_configured_provider_name(
content_type="combined",
user_id=db_user_id,
fallback_to_main=False,
)
providers = []
for info in list_providers():
enabled_key = f"{info['name'].upper()}_ENABLED"
provider_info = {
"name": info["name"],
"display_name": info["display_name"],
"requires_auth": info["requires_auth"],
"enabled": app_config.get(enabled_key, False) is True,
"available": False,
}
try:
kwargs = get_provider_kwargs(info["name"])
provider = get_provider(info["name"], **kwargs)
provider_info["available"] = provider.is_available()
except _OPERATIONAL_ERRORS as exc:
logger.debug(
"Metadata provider %s availability check failed: %s",
info["name"],
exc,
)
providers.append(provider_info)
return jsonify(
{
"providers": providers,
"configured_provider": configured_metadata_provider or None,
"configured_provider_audiobook": configured_audiobook_metadata_provider or None,
"configured_provider_combined": configured_combined_metadata_provider or None,
}
)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata providers error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/config", methods=["GET"])
@login_required
def api_metadata_config() -> Response | tuple[Response, int]:
"""Return provider-specific metadata search config for the active session."""
try:
from shelfmark.metadata_providers import (
get_configured_provider_name,
get_provider,
get_provider_capabilities,
get_provider_default_sort,
get_provider_kwargs,
get_provider_search_fields,
get_provider_sort_options,
is_provider_registered,
)
app_config.refresh()
content_type = request.args.get("content_type", "ebook").strip()
provider_name = request.args.get("provider", "").strip()
db_user_id = get_session_db_user_id(session)
if not provider_name:
provider_name = get_configured_provider_name(
content_type=content_type,
user_id=db_user_id,
fallback_to_main=True,
)
if not provider_name:
return jsonify(
{
"provider": None,
"display_name": None,
"enabled": False,
"available": False,
"search_fields": [],
"capabilities": [],
"sort_options": [{"value": "relevance", "label": "Most relevant"}],
"default_sort": "relevance",
}
)
if not is_provider_registered(provider_name):
return jsonify({"error": f"Unknown metadata provider: {provider_name}"}), 400
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
enabled_key = f"{provider_name.upper()}_ENABLED"
provider_enabled = app_config.get(enabled_key, False) is True
provider_available = provider.is_available()
return jsonify(
{
"provider": provider_name,
"display_name": provider.display_name,
"enabled": provider_enabled,
"available": provider_available,
"search_fields": get_provider_search_fields(provider_name),
"capabilities": get_provider_capabilities(provider_name),
"sort_options": get_provider_sort_options(provider_name),
"default_sort": get_provider_default_sort(provider_name, user_id=db_user_id),
}
)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata config error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/search", methods=["GET"])
@login_required
def api_metadata_search() -> Response | tuple[Response, int]:
"""Search for books using the configured metadata provider.
Query Parameters:
query (str): Search query (required)
limit (int): Maximum number of results (default: 40, max: 100)
sort (str): Sort order - relevance, popularity, rating, newest, oldest (default: relevance)
[dynamic fields]: Provider-specific search fields passed as query params
Returns:
flask.Response: JSON with list of books from metadata provider.
"""
try:
from dataclasses import asdict
from shelfmark.metadata_providers import (
CheckboxSearchField,
MetadataSearchOptions,
NumberSearchField,
SortOrder,
get_configured_provider,
get_provider,
get_provider_kwargs,
is_provider_enabled,
is_provider_registered,
)
query = request.args.get("query", "").strip()
content_type = request.args.get("content_type", "ebook").strip()
provider_name = request.args.get("provider", "").strip()
try:
limit = min(int(request.args.get("limit", 40)), 100)
except ValueError:
limit = 40
try:
page = max(1, int(request.args.get("page", 1)))
except ValueError:
page = 1
# Parse sort parameter
sort_value = request.args.get("sort", "relevance").lower()
try:
sort_order = SortOrder(sort_value)
except ValueError:
sort_order = SortOrder.RELEVANCE
db_user_id = get_session_db_user_id(session)
if provider_name:
if not is_provider_registered(provider_name):
return jsonify(
{
"error": f"Unknown metadata provider: {provider_name}",
"message": f"Unknown metadata provider: {provider_name}",
}
), 400
if not is_provider_enabled(provider_name):
return jsonify(
{
"error": f"Metadata provider '{provider_name}' is not enabled",
"message": f"{provider_name} is not enabled. Enable it in Settings first.",
}
), 503
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
else:
provider = get_configured_provider(content_type=content_type, user_id=db_user_id)
if not provider:
return jsonify(
{
"error": "No metadata provider configured",
"message": "No metadata provider configured. Enable one in Settings.",
}
), 503
if not provider.is_available():
return jsonify(
{
"error": f"Metadata provider '{provider.name}' is not available",
"message": f"{provider.display_name} is not available. Check configuration in Settings.",
}
), 503
# Extract custom search field values from query params
fields: dict[str, Any] = {}
may_use_account = _session_may_use_provider_account()
for search_field in provider.search_fields:
if search_field.key in provider.account_search_fields and not may_use_account:
continue
value = request.args.get(search_field.key)
if value is not None:
# Strip string values to handle whitespace-only input
value = value.strip()
if value != "":
# Parse value based on field type
if isinstance(search_field, CheckboxSearchField):
fields[search_field.key] = value.lower() in ("true", "1", "yes", "on")
elif isinstance(search_field, NumberSearchField):
with suppress(ValueError):
fields[search_field.key] = int(value)
else:
fields[search_field.key] = value
# Require either a query or at least one field value
if not query and not fields:
return jsonify({"error": "Either 'query' or search field values are required"}), 400
options = MetadataSearchOptions(
query=query,
limit=limit,
page=page,
sort=sort_order,
fields=fields,
user_id=db_user_id,
)
search_result = provider.search_paginated(options)
# Convert BookMetadata objects to dicts
books_data = [asdict(book) for book in search_result.books]
# Transform cover_url to local proxy URLs when caching is enabled
from shelfmark.core.utils import transform_cover_url
for book_dict in books_data:
if book_dict.get("cover_url"):
cache_id = f"{book_dict['provider']}_{book_dict['provider_id']}"
book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
for book, book_dict in zip(search_result.books, books_data, strict=True):
attach_library_ownership(book, book_dict)
response_data = {
"books": books_data,
"provider": provider.name,
"query": query,
"page": search_result.page,
"total_found": search_result.total_found,
"has_more": search_result.has_more,
}
if search_result.source_url:
response_data["source_url"] = search_result.source_url
if search_result.source_title:
response_data["source_title"] = search_result.source_title
return jsonify(response_data)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata search error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/field-options", methods=["GET"])
@login_required
def api_metadata_field_options() -> Response:
"""Return dynamic search-field options for a metadata provider."""
try:
from shelfmark.metadata_providers import (
get_configured_provider,
get_provider,
get_provider_kwargs,
is_provider_registered,
)
field_key = request.args.get("field", "").strip()
provider_name = request.args.get("provider", "").strip()
content_type = request.args.get("content_type", "ebook").strip()
query_text = request.args.get("query", "").strip()
if not field_key:
return jsonify({"options": []})
db_user_id = get_session_db_user_id(session)
provider = None
if provider_name:
if not is_provider_registered(provider_name):
return jsonify({"options": []})
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
else:
provider = get_configured_provider(content_type=content_type, user_id=db_user_id)
if not provider or not provider.is_available():
return jsonify({"options": []})
if (
field_key in provider.account_search_fields
and not _session_may_use_provider_account()
):
return jsonify({"options": []})
options = provider.get_search_field_options(field_key, query=query_text or None)
return jsonify({"options": options})
except _OPERATIONAL_ERRORS as e:
logger.warning("Metadata field options endpoint error: %s", e)
return jsonify({"options": []})
@app.route("/api/metadata/book/<provider>/<book_id>", methods=["GET"])
@login_required
def api_metadata_book(provider: str, book_id: str) -> Response | tuple[Response, int]:
"""Get detailed book information from a metadata provider.
Path Parameters:
provider (str): Provider name (e.g., "hardcover", "openlibrary")
book_id (str): Book ID in the provider's system
Returns:
flask.Response: JSON with book details.
"""
try:
from dataclasses import asdict
prov = _resolve_metadata_provider(provider)
book = prov.get_book(book_id)
if not book:
return jsonify({"error": "Book not found"}), 404
book_dict = asdict(book)
attach_library_ownership(book, book_dict)
# Transform cover_url to local proxy URL when caching is enabled
from shelfmark.core.utils import transform_cover_url
if book_dict.get("cover_url"):
cache_id = f"{provider}_{book_id}"
book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
return jsonify(book_dict)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except RuntimeError as e:
return jsonify({"error": str(e)}), 503
except (OSError, TypeError, sqlite3.Error) as e:
logger.error_trace(f"Metadata book error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/book/<provider>/<book_id>/targets", methods=["GET"])
@login_required
@_handle_target_errors("Failed to load book targets")
def api_metadata_book_targets(provider: str, book_id: str) -> Response | tuple[Response, int]:
"""Get provider-managed list/status targets for a specific book."""
prov = _resolve_metadata_provider(provider)
if not _session_may_use_provider_account():
return jsonify({"options": []})
return jsonify({"options": prov.get_book_targets(book_id)})
@app.route("/api/metadata/book/<provider>/targets/batch", methods=["POST"])
@login_required
@_handle_target_errors("Failed to load book targets")
def api_metadata_book_targets_batch(provider: str) -> Response | tuple[Response, int]:
"""Get provider-managed list/status targets for multiple books."""
prov = _resolve_metadata_provider(provider)
payload = request.get_json(silent=True) or {}
raw_ids = payload.get("book_ids", []) if isinstance(payload, dict) else []
if not isinstance(raw_ids, list) or not raw_ids:
return jsonify({"error": "book_ids must be a non-empty array"}), 400
book_ids = [str(bid) for bid in raw_ids[:50]]
if not _session_may_use_provider_account():
return jsonify({"results": {book_id: [] for book_id in book_ids}})
return jsonify({"results": prov.get_book_targets_batch(book_ids)})
@app.route("/api/metadata/book/<provider>/<book_id>/targets", methods=["PUT"])
@login_required
@_handle_target_errors("Failed to update book targets")
def api_metadata_book_targets_update(
provider: str, book_id: str
) -> Response | tuple[Response, int]:
"""Set whether a book belongs to a provider-managed list or shelf."""
if not _session_may_use_provider_account():
return jsonify({"error": "Admin access required"}), 403
prov = _resolve_metadata_provider(provider)
payload = request.get_json(silent=True) or {}
target = str(payload.get("target", "")).strip() if isinstance(payload, dict) else ""
selected = payload.get("selected") if isinstance(payload, dict) else None
if not target:
return jsonify({"error": "target is required"}), 400
if not isinstance(selected, bool):
return jsonify({"error": "selected must be a boolean"}), 400
result = prov.set_book_target_state(book_id, target, selected=selected)
response: dict = {
"success": True,
"changed": bool(result.get("changed", True)),
"selected": selected,
}
deselected = result.get("deselected_target")
if isinstance(deselected, str) and deselected:
response["deselected_target"] = deselected
return jsonify(response)
+8 -500
View File
@@ -7,7 +7,6 @@ import re
import sqlite3
import time
import unicodedata
from contextlib import suppress
from datetime import UTC, datetime, timedelta
from functools import wraps
from importlib import import_module
@@ -58,6 +57,7 @@ from shelfmark.core.cwa_user_sync import upsert_cwa_user
from shelfmark.core.download_history_service import DownloadHistoryService
from shelfmark.core.external_user_linking import upsert_external_user
from shelfmark.core.logger import setup_logger
from shelfmark.core.metadata_routes import register_metadata_routes
from shelfmark.core.models import (
TERMINAL_QUEUE_STATUSES,
QueueStatus,
@@ -118,7 +118,7 @@ from shelfmark.release_sources import (
if TYPE_CHECKING:
from shelfmark.core.models import DownloadTask
from shelfmark.metadata_providers import BookMetadata, MetadataProvider
from shelfmark.metadata_providers import BookMetadata
logger = setup_logger(__name__)
FLASK_SECRET_KEY_MIN_BYTES = 32
@@ -2726,341 +2726,6 @@ def api_auth_check() -> Response | tuple[Response, int]:
)
@app.route("/api/metadata/providers", methods=["GET"])
@login_required
def api_metadata_providers() -> Response | tuple[Response, int]:
"""Get list of available metadata providers.
Returns:
flask.Response: JSON with list of providers and their status.
"""
try:
from shelfmark.metadata_providers import (
get_configured_provider_name,
get_provider,
get_provider_kwargs,
list_providers,
)
app_config.refresh()
db_user_id = get_session_db_user_id(session)
configured_metadata_provider = get_configured_provider_name(
content_type="ebook",
user_id=db_user_id,
fallback_to_main=True,
)
configured_audiobook_metadata_provider = get_configured_provider_name(
content_type="audiobook",
user_id=db_user_id,
fallback_to_main=False,
)
configured_combined_metadata_provider = get_configured_provider_name(
content_type="combined",
user_id=db_user_id,
fallback_to_main=False,
)
providers = []
for info in list_providers():
enabled_key = f"{info['name'].upper()}_ENABLED"
provider_info = {
"name": info["name"],
"display_name": info["display_name"],
"requires_auth": info["requires_auth"],
"enabled": app_config.get(enabled_key, False) is True,
"available": False,
}
try:
kwargs = get_provider_kwargs(info["name"])
provider = get_provider(info["name"], **kwargs)
provider_info["available"] = provider.is_available()
except _OPERATIONAL_ERRORS as exc:
logger.debug(
"Metadata provider %s availability check failed: %s",
info["name"],
exc,
)
providers.append(provider_info)
return jsonify(
{
"providers": providers,
"configured_provider": configured_metadata_provider or None,
"configured_provider_audiobook": configured_audiobook_metadata_provider or None,
"configured_provider_combined": configured_combined_metadata_provider or None,
}
)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata providers error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/config", methods=["GET"])
@login_required
def api_metadata_config() -> Response | tuple[Response, int]:
"""Return provider-specific metadata search config for the active session."""
try:
from shelfmark.metadata_providers import (
get_configured_provider_name,
get_provider,
get_provider_capabilities,
get_provider_default_sort,
get_provider_kwargs,
get_provider_search_fields,
get_provider_sort_options,
is_provider_registered,
)
app_config.refresh()
content_type = request.args.get("content_type", "ebook").strip()
provider_name = request.args.get("provider", "").strip()
db_user_id = get_session_db_user_id(session)
if not provider_name:
provider_name = get_configured_provider_name(
content_type=content_type,
user_id=db_user_id,
fallback_to_main=True,
)
if not provider_name:
return jsonify(
{
"provider": None,
"display_name": None,
"enabled": False,
"available": False,
"search_fields": [],
"capabilities": [],
"sort_options": [{"value": "relevance", "label": "Most relevant"}],
"default_sort": "relevance",
}
)
if not is_provider_registered(provider_name):
return jsonify({"error": f"Unknown metadata provider: {provider_name}"}), 400
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
enabled_key = f"{provider_name.upper()}_ENABLED"
provider_enabled = app_config.get(enabled_key, False) is True
provider_available = provider.is_available()
return jsonify(
{
"provider": provider_name,
"display_name": provider.display_name,
"enabled": provider_enabled,
"available": provider_available,
"search_fields": get_provider_search_fields(provider_name),
"capabilities": get_provider_capabilities(provider_name),
"sort_options": get_provider_sort_options(provider_name),
"default_sort": get_provider_default_sort(provider_name, user_id=db_user_id),
}
)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata config error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/search", methods=["GET"])
@login_required
def api_metadata_search() -> Response | tuple[Response, int]:
"""Search for books using the configured metadata provider.
Query Parameters:
query (str): Search query (required)
limit (int): Maximum number of results (default: 40, max: 100)
sort (str): Sort order - relevance, popularity, rating, newest, oldest (default: relevance)
[dynamic fields]: Provider-specific search fields passed as query params
Returns:
flask.Response: JSON with list of books from metadata provider.
"""
try:
from dataclasses import asdict
from shelfmark.metadata_providers import (
CheckboxSearchField,
MetadataSearchOptions,
NumberSearchField,
SortOrder,
get_configured_provider,
get_provider,
get_provider_kwargs,
is_provider_enabled,
is_provider_registered,
)
query = request.args.get("query", "").strip()
content_type = request.args.get("content_type", "ebook").strip()
provider_name = request.args.get("provider", "").strip()
try:
limit = min(int(request.args.get("limit", 40)), 100)
except ValueError:
limit = 40
try:
page = max(1, int(request.args.get("page", 1)))
except ValueError:
page = 1
# Parse sort parameter
sort_value = request.args.get("sort", "relevance").lower()
try:
sort_order = SortOrder(sort_value)
except ValueError:
sort_order = SortOrder.RELEVANCE
db_user_id = get_session_db_user_id(session)
if provider_name:
if not is_provider_registered(provider_name):
return jsonify(
{
"error": f"Unknown metadata provider: {provider_name}",
"message": f"Unknown metadata provider: {provider_name}",
}
), 400
if not is_provider_enabled(provider_name):
return jsonify(
{
"error": f"Metadata provider '{provider_name}' is not enabled",
"message": f"{provider_name} is not enabled. Enable it in Settings first.",
}
), 503
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
else:
provider = get_configured_provider(content_type=content_type, user_id=db_user_id)
if not provider:
return jsonify(
{
"error": "No metadata provider configured",
"message": "No metadata provider configured. Enable one in Settings.",
}
), 503
if not provider.is_available():
return jsonify(
{
"error": f"Metadata provider '{provider.name}' is not available",
"message": f"{provider.display_name} is not available. Check configuration in Settings.",
}
), 503
# Extract custom search field values from query params
fields: dict[str, Any] = {}
may_use_account = _session_may_use_provider_account()
for search_field in provider.search_fields:
if search_field.key in provider.account_search_fields and not may_use_account:
continue
value = request.args.get(search_field.key)
if value is not None:
# Strip string values to handle whitespace-only input
value = value.strip()
if value != "":
# Parse value based on field type
if isinstance(search_field, CheckboxSearchField):
fields[search_field.key] = value.lower() in ("true", "1", "yes", "on")
elif isinstance(search_field, NumberSearchField):
with suppress(ValueError):
fields[search_field.key] = int(value)
else:
fields[search_field.key] = value
# Require either a query or at least one field value
if not query and not fields:
return jsonify({"error": "Either 'query' or search field values are required"}), 400
options = MetadataSearchOptions(
query=query, limit=limit, page=page, sort=sort_order, fields=fields, user_id=db_user_id
)
search_result = provider.search_paginated(options)
# Convert BookMetadata objects to dicts
books_data = [asdict(book) for book in search_result.books]
# Transform cover_url to local proxy URLs when caching is enabled
from shelfmark.core.utils import transform_cover_url
for book_dict in books_data:
if book_dict.get("cover_url"):
cache_id = f"{book_dict['provider']}_{book_dict['provider_id']}"
book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
for book, book_dict in zip(search_result.books, books_data, strict=True):
_attach_library_ownership(book, book_dict)
response_data = {
"books": books_data,
"provider": provider.name,
"query": query,
"page": search_result.page,
"total_found": search_result.total_found,
"has_more": search_result.has_more,
}
if search_result.source_url:
response_data["source_url"] = search_result.source_url
if search_result.source_title:
response_data["source_title"] = search_result.source_title
return jsonify(response_data)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Metadata search error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/metadata/field-options", methods=["GET"])
@login_required
def api_metadata_field_options() -> Response:
"""Return dynamic search-field options for a metadata provider."""
try:
from shelfmark.metadata_providers import (
get_configured_provider,
get_provider,
get_provider_kwargs,
is_provider_registered,
)
field_key = request.args.get("field", "").strip()
provider_name = request.args.get("provider", "").strip()
content_type = request.args.get("content_type", "ebook").strip()
query_text = request.args.get("query", "").strip()
if not field_key:
return jsonify({"options": []})
db_user_id = get_session_db_user_id(session)
provider = None
if provider_name:
if not is_provider_registered(provider_name):
return jsonify({"options": []})
kwargs = get_provider_kwargs(provider_name)
provider = get_provider(provider_name, **kwargs)
else:
provider = get_configured_provider(content_type=content_type, user_id=db_user_id)
if not provider or not provider.is_available():
return jsonify({"options": []})
if field_key in provider.account_search_fields and not _session_may_use_provider_account():
return jsonify({"options": []})
options = provider.get_search_field_options(field_key, query=query_text or None)
return jsonify({"options": options})
except _OPERATIONAL_ERRORS as e:
logger.warning("Metadata field options endpoint error: %s", e)
return jsonify({"options": []})
def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None:
"""Add per-format ownership under ``library`` when a library check is enabled."""
from shelfmark.core import library_index
@@ -3070,169 +2735,12 @@ def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None:
book_dict["library"] = owned
def _session_may_use_provider_account() -> bool:
"""Whether this session may read or change the metadata provider's own account.
A provider such as Hardcover is signed in as one account, the instance owner's, and its lists
and shelves belong to that person. With no authentication configured the one user is the owner.
Otherwise only an admin may use them.
"""
return get_auth_mode() == "none" or bool(session.get("is_admin", False))
def _resolve_metadata_provider(provider_name: str) -> MetadataProvider:
"""Validate, instantiate and return a ready metadata provider.
Raises appropriate HTTP-friendly exceptions on failure.
"""
from shelfmark.metadata_providers import (
get_provider,
get_provider_kwargs,
is_provider_registered,
)
if not is_provider_registered(provider_name):
msg = f"Unknown metadata provider: {provider_name}"
raise ValueError(msg)
kwargs = get_provider_kwargs(provider_name)
prov = get_provider(provider_name, **kwargs)
if not prov.is_available():
msg = f"Provider '{provider_name}' is not available"
raise RuntimeError(msg)
return prov
@app.route("/api/metadata/book/<provider>/<book_id>", methods=["GET"])
@login_required
def api_metadata_book(provider: str, book_id: str) -> Response | tuple[Response, int]:
"""Get detailed book information from a metadata provider.
Path Parameters:
provider (str): Provider name (e.g., "hardcover", "openlibrary")
book_id (str): Book ID in the provider's system
Returns:
flask.Response: JSON with book details.
"""
try:
from dataclasses import asdict
prov = _resolve_metadata_provider(provider)
book = prov.get_book(book_id)
if not book:
return jsonify({"error": "Book not found"}), 404
book_dict = asdict(book)
_attach_library_ownership(book, book_dict)
# Transform cover_url to local proxy URL when caching is enabled
from shelfmark.core.utils import transform_cover_url
if book_dict.get("cover_url"):
cache_id = f"{provider}_{book_id}"
book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id)
return jsonify(book_dict)
except ValueError as e:
return jsonify({"error": str(e)}), 400
except RuntimeError as e:
return jsonify({"error": str(e)}), 503
except (OSError, TypeError, sqlite3.Error) as e:
logger.error_trace(f"Metadata book error: {e}")
return jsonify({"error": str(e)}), 500
def _handle_target_errors(
fallback_message: str,
) -> Callable[
[Callable[..., Response | tuple[Response, int]]], Callable[..., Response | tuple[Response, int]]
]:
"""Wrap a metadata-target route with standard error handling."""
def decorator(
fn: Callable[..., Response | tuple[Response, int]],
) -> Callable[..., Response | tuple[Response, int]]:
@wraps(fn)
def wrapper(*args: object, **kwargs: object) -> Response | tuple[Response, int]:
try:
return fn(*args, **kwargs)
except (NotImplementedError, ValueError) as e:
return jsonify({"error": str(e)}), 400
except RuntimeError as e:
return jsonify({"error": str(e)}), 502
except (OSError, TypeError, sqlite3.Error) as e:
logger.error_trace(f"{fallback_message}: {e}")
return jsonify({"error": fallback_message}), 500
return wrapper
return decorator
@app.route("/api/metadata/book/<provider>/<book_id>/targets", methods=["GET"])
@login_required
@_handle_target_errors("Failed to load book targets")
def api_metadata_book_targets(provider: str, book_id: str) -> Response | tuple[Response, int]:
"""Get provider-managed list/status targets for a specific book."""
prov = _resolve_metadata_provider(provider)
if not _session_may_use_provider_account():
return jsonify({"options": []})
return jsonify({"options": prov.get_book_targets(book_id)})
@app.route("/api/metadata/book/<provider>/targets/batch", methods=["POST"])
@login_required
@_handle_target_errors("Failed to load book targets")
def api_metadata_book_targets_batch(provider: str) -> Response | tuple[Response, int]:
"""Get provider-managed list/status targets for multiple books."""
prov = _resolve_metadata_provider(provider)
payload = request.get_json(silent=True) or {}
raw_ids = payload.get("book_ids", []) if isinstance(payload, dict) else []
if not isinstance(raw_ids, list) or not raw_ids:
return jsonify({"error": "book_ids must be a non-empty array"}), 400
book_ids = [str(bid) for bid in raw_ids[:50]]
if not _session_may_use_provider_account():
return jsonify({"results": {book_id: [] for book_id in book_ids}})
return jsonify({"results": prov.get_book_targets_batch(book_ids)})
@app.route("/api/metadata/book/<provider>/<book_id>/targets", methods=["PUT"])
@login_required
@_handle_target_errors("Failed to update book targets")
def api_metadata_book_targets_update(
provider: str, book_id: str
) -> Response | tuple[Response, int]:
"""Set whether a book belongs to a provider-managed list or shelf."""
if not _session_may_use_provider_account():
return jsonify({"error": "Admin access required"}), 403
prov = _resolve_metadata_provider(provider)
payload = request.get_json(silent=True) or {}
target = str(payload.get("target", "")).strip() if isinstance(payload, dict) else ""
selected = payload.get("selected") if isinstance(payload, dict) else None
if not target:
return jsonify({"error": "target is required"}), 400
if not isinstance(selected, bool):
return jsonify({"error": "selected must be a boolean"}), 400
result = prov.set_book_target_state(book_id, target, selected=selected)
response: dict = {
"success": True,
"changed": bool(result.get("changed", True)),
"selected": selected,
}
deselected = result.get("deselected_target")
if isinstance(deselected, str) and deselected:
response["deselected_target"] = deselected
return jsonify(response)
register_metadata_routes(
app,
login_required=login_required,
resolve_auth_mode=_resolve_auth_mode_for_routes,
attach_library_ownership=_attach_library_ownership,
)
@app.route("/api/releases", methods=["GET"])
+348
View File
@@ -0,0 +1,348 @@
"""Behaviour of the `/api/metadata/...` routes that no other test pins down."""
from __future__ import annotations
import importlib
from unittest.mock import patch
import pytest
from shelfmark.metadata_providers import BookMetadata
@pytest.fixture(scope="module")
def main_module():
"""Import `shelfmark.main` with background startup disabled."""
with patch("shelfmark.download.orchestrator.start"):
import shelfmark.main as main
importlib.reload(main)
return main
class _Provider:
name = "stub"
display_name = "Stub"
search_fields: list = []
account_search_fields: frozenset = frozenset()
def __init__(self) -> None:
self.available = True
self.book: BookMetadata | None = BookMetadata(
provider="stub", provider_id="7", title="Dune", authors=["Frank Herbert"]
)
self.raise_on_targets: Exception | None = None
def is_available(self) -> bool:
return self.available
def get_book(self, book_id: str) -> BookMetadata | None:
return self.book
def get_book_targets(self, book_id: str) -> list:
if self.raise_on_targets:
raise self.raise_on_targets
return [{"value": "list:1"}]
def get_book_targets_batch(self, ids: list) -> dict:
return {i: [] for i in ids}
def set_book_target_state(self, book_id: str, target: str, *, selected: bool) -> dict:
return {"changed": False, "deselected_target": "list:2"}
@pytest.fixture
def stub() -> _Provider:
return _Provider()
@pytest.fixture
def client(main_module, stub):
registered = {"stub": True}
with (
patch("shelfmark.metadata_providers.get_provider", return_value=stub),
patch(
"shelfmark.metadata_providers.is_provider_registered",
side_effect=lambda name: registered.get(name, False),
),
patch("shelfmark.metadata_providers.get_provider_kwargs", return_value={}),
patch.object(main_module, "get_auth_mode", return_value="none"),
):
yield main_module.app.test_client()
class TestAuth:
@pytest.mark.parametrize(
("method", "path"),
[
("get", "/api/metadata/providers"),
("get", "/api/metadata/config"),
("get", "/api/metadata/search?query=x"),
("get", "/api/metadata/field-options?field=a"),
("get", "/api/metadata/book/stub/7"),
("get", "/api/metadata/book/stub/7/targets"),
("post", "/api/metadata/book/stub/targets/batch"),
("put", "/api/metadata/book/stub/7/targets"),
],
)
def test_every_route_needs_a_session(self, main_module, method, path) -> None:
client = main_module.app.test_client()
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
resp = getattr(client, method)(path)
assert resp.status_code == 401
class TestProviders:
def test_lists_providers_with_availability_and_configured_names(self, client, stub) -> None:
stub.available = False
info = {"name": "stub", "display_name": "Stub", "requires_auth": True}
with (
patch("shelfmark.metadata_providers.list_providers", return_value=[info]),
patch(
"shelfmark.metadata_providers.get_configured_provider_name",
side_effect=lambda content_type, **_: {"ebook": "stub", "audiobook": ""}.get(
content_type, ""
),
),
):
resp = client.get("/api/metadata/providers")
assert resp.status_code == 200
assert resp.json == {
"providers": [
{
"name": "stub",
"display_name": "Stub",
"requires_auth": True,
"enabled": False,
"available": False,
}
],
"configured_provider": "stub",
"configured_provider_audiobook": None,
"configured_provider_combined": None,
}
def test_a_provider_that_fails_its_check_is_listed_unavailable(self, client) -> None:
info = {"name": "stub", "display_name": "Stub", "requires_auth": False}
with (
patch("shelfmark.metadata_providers.list_providers", return_value=[info]),
patch("shelfmark.metadata_providers.get_provider", side_effect=RuntimeError("boom")),
patch("shelfmark.metadata_providers.get_configured_provider_name", return_value=""),
):
resp = client.get("/api/metadata/providers")
assert resp.status_code == 200
assert resp.json["providers"][0]["available"] is False
class TestConfig:
def test_no_provider_configured_returns_the_empty_shape(self, client) -> None:
with patch("shelfmark.metadata_providers.get_configured_provider_name", return_value=""):
resp = client.get("/api/metadata/config")
assert resp.status_code == 200
assert resp.json["provider"] is None
assert resp.json["search_fields"] == []
assert resp.json["default_sort"] == "relevance"
def test_unknown_provider_is_a_400(self, client) -> None:
resp = client.get("/api/metadata/config?provider=nope")
assert resp.status_code == 400
assert resp.json == {"error": "Unknown metadata provider: nope"}
def test_known_provider_returns_its_search_config(self, client) -> None:
with (
patch("shelfmark.metadata_providers.get_provider_search_fields", return_value=[1]),
patch("shelfmark.metadata_providers.get_provider_capabilities", return_value=["c"]),
patch("shelfmark.metadata_providers.get_provider_sort_options", return_value=[2]),
patch("shelfmark.metadata_providers.get_provider_default_sort", return_value="newest"),
):
resp = client.get("/api/metadata/config?provider=stub")
assert resp.json == {
"provider": "stub",
"display_name": "Stub",
"enabled": False,
"available": True,
"search_fields": [1],
"capabilities": ["c"],
"sort_options": [2],
"default_sort": "newest",
}
class TestSearchErrors:
def test_unknown_provider_is_a_400(self, client) -> None:
resp = client.get("/api/metadata/search?provider=nope&query=x")
assert resp.status_code == 400
assert resp.json["error"] == "Unknown metadata provider: nope"
def test_disabled_provider_is_a_503(self, client) -> None:
with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=False):
resp = client.get("/api/metadata/search?provider=stub&query=x")
assert resp.status_code == 503
assert resp.json["error"] == "Metadata provider 'stub' is not enabled"
def test_no_configured_provider_is_a_503(self, client) -> None:
with patch("shelfmark.metadata_providers.get_configured_provider", return_value=None):
resp = client.get("/api/metadata/search?query=x")
assert resp.status_code == 503
assert resp.json["error"] == "No metadata provider configured"
def test_unavailable_provider_is_a_503(self, client, stub) -> None:
stub.available = False
with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True):
resp = client.get("/api/metadata/search?provider=stub&query=x")
assert resp.status_code == 503
assert resp.json["error"] == "Metadata provider 'stub' is not available"
def test_no_query_and_no_fields_is_a_400(self, client) -> None:
with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True):
resp = client.get("/api/metadata/search?provider=stub")
assert resp.status_code == 400
assert resp.json == {"error": "Either 'query' or search field values are required"}
class TestFieldOptions:
def test_no_field_gives_no_options(self, client) -> None:
assert client.get("/api/metadata/field-options").json == {"options": []}
def test_unknown_provider_gives_no_options(self, client) -> None:
resp = client.get("/api/metadata/field-options?provider=nope&field=a")
assert resp.json == {"options": []}
def test_unavailable_provider_gives_no_options(self, client, stub) -> None:
stub.available = False
assert client.get("/api/metadata/field-options?provider=stub&field=a").json == {
"options": []
}
class TestBook:
def test_returns_the_book(self, client) -> None:
resp = client.get("/api/metadata/book/stub/7")
assert resp.status_code == 200
assert resp.json["title"] == "Dune"
assert resp.json["provider_id"] == "7"
def test_missing_book_is_a_404(self, client, stub) -> None:
stub.book = None
resp = client.get("/api/metadata/book/stub/7")
assert resp.status_code == 404
assert resp.json == {"error": "Book not found"}
def test_unknown_provider_is_a_400(self, client) -> None:
resp = client.get("/api/metadata/book/nope/7")
assert resp.status_code == 400
assert resp.json == {"error": "Unknown metadata provider: nope"}
def test_unavailable_provider_is_a_503(self, client, stub) -> None:
stub.available = False
resp = client.get("/api/metadata/book/stub/7")
assert resp.status_code == 503
assert resp.json == {"error": "Provider 'stub' is not available"}
def test_a_library_ownership_check_adds_the_library_key(self, client) -> None:
with patch("shelfmark.core.library_index.ownership", return_value={"epub": True}):
resp = client.get("/api/metadata/book/stub/7")
assert resp.json["library"] == {"epub": True}
class TestTargetErrors:
def test_unknown_provider_is_a_400(self, client) -> None:
resp = client.get("/api/metadata/book/nope/7/targets")
assert resp.status_code == 400
def test_unavailable_provider_is_a_502(self, client, stub) -> None:
stub.available = False
resp = client.get("/api/metadata/book/stub/7/targets")
assert resp.status_code == 502
assert resp.json == {"error": "Provider 'stub' is not available"}
def test_not_implemented_is_a_400(self, client, stub) -> None:
stub.raise_on_targets = NotImplementedError("no targets here")
resp = client.get("/api/metadata/book/stub/7/targets")
assert resp.status_code == 400
assert resp.json == {"error": "no targets here"}
def test_an_operational_error_gets_the_fallback_message(self, client, stub) -> None:
stub.raise_on_targets = OSError("secret detail")
resp = client.get("/api/metadata/book/stub/7/targets")
assert resp.status_code == 500
assert resp.json == {"error": "Failed to load book targets"}
def test_the_update_route_uses_its_own_fallback_message(self, client, stub) -> None:
with patch.object(stub, "set_book_target_state", side_effect=TypeError("x")):
resp = client.put(
"/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": True}
)
assert resp.status_code == 500
assert resp.json == {"error": "Failed to update book targets"}
class TestTargetPayloads:
@pytest.mark.parametrize("body", [{}, {"book_ids": []}, {"book_ids": "x"}, []])
def test_batch_needs_a_non_empty_list(self, client, body) -> None:
resp = client.post("/api/metadata/book/stub/targets/batch", json=body)
assert resp.status_code == 400
assert resp.json == {"error": "book_ids must be a non-empty array"}
def test_batch_caps_the_ids_at_fifty_and_stringifies_them(self, client) -> None:
resp = client.post(
"/api/metadata/book/stub/targets/batch", json={"book_ids": list(range(60))}
)
assert resp.status_code == 200
assert set(resp.json["results"]) == {str(i) for i in range(50)}
def test_update_needs_a_target(self, client) -> None:
resp = client.put("/api/metadata/book/stub/7/targets", json={"selected": True})
assert resp.status_code == 400
assert resp.json == {"error": "target is required"}
def test_update_needs_a_boolean_selected(self, client) -> None:
resp = client.put(
"/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": "yes"}
)
assert resp.status_code == 400
assert resp.json == {"error": "selected must be a boolean"}
def test_update_reports_the_change_and_the_deselected_target(self, client) -> None:
resp = client.put(
"/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": True}
)
assert resp.json == {
"success": True,
"changed": False,
"selected": True,
"deselected_target": "list:2",
}