migrate /api/books and /api/volumes to fastapi (#11922)

Author: RayBB

openlibrary/asgi_app.py

@@ -211,6 +211,7 @@ def health() -> dict[str, str]:
         return {"status": "ok"}
 
     from openlibrary.fastapi.account import router as account_router
+    from openlibrary.fastapi.books import router as books_router
     from openlibrary.fastapi.cdn import router as cdn_router
     from openlibrary.fastapi.checkins import router as checkins_router
     from openlibrary.fastapi.internal.api import router as internal_router
@@ -227,6 +228,7 @@ def health() -> dict[str, str]:
 
     # Include routers
     app.include_router(account_router)
+    app.include_router(books_router)
     app.include_router(cdn_router)
     app.include_router(checkins_router)
     app.include_router(internal_router)

openlibrary/fastapi/books.py

@@ -0,0 +1,156 @@
+"""FastAPI Books API endpoints.
+
+DO NOT follow the example of this code... it is horrible and not following
+the best practices like Response Models because JSONP makes it quite complicated.
+"""
+
+from __future__ import annotations
+
+import re
+import urllib.parse
+from typing import Annotated, Any, Literal
+
+import web
+from fastapi import APIRouter, Query, Request
+from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter
+
+from openlibrary.fastapi.models import parse_comma_separated_list, wrap_jsonp
+from openlibrary.plugins.books import dynlinks, readlinks
+from openlibrary.plugins.books.dynlinks import DynlinksOptions
+
+router = APIRouter()
+
+
+class BooksAPIQueryParams(BaseModel):
+    """Query parameters for Books API endpoint."""
+
+    bibkeys: Annotated[list[str], BeforeValidator(parse_comma_separated_list)] = Field(
+        ...,
+        description="Comma-separated list of bibliography keys (ISBN, LCCN, OCLC, etc.)",
+    )
+    details: Literal["true", "false"] = Field("false", description="Include detailed book information")
+    jscmd: Literal["details", "data", "viewapi"] | None = Field(None, description="Format of returned data")
+    callback: str | None = Field(None, description="JSONP callback function name")
+    high_priority: bool = Field(False, description="Attempt import immediately for missing ISBNs")
+    format: Literal["json", "js"] | None = Field(None, description="Explicitly set response format (overrides path-based detection)")
+
+
+# Note: We don't set response_model on these endpoints because they support JSONP callback.
+# When callback parameter is present, the function returns a Response object directly,
+# which should bypass response model validation. Setting response_model on the decorator
+# can interfere with Response object's content-type headers.
+@router.get("/api/books", include_in_schema=False)
+@router.get("/api/books.json")
+async def get_books(
+    request: Request,
+    params: Annotated[BooksAPIQueryParams, Query()],
+) -> Any:
+    """
+    Get book metadata by bibliography keys.
+
+    This endpoint provides basic information about books, including URLs,
+    thumbnail links, and preview availability. Supports ISBNs, LCCNs,
+    OCLC numbers, and Open Library IDs.
+
+    When `high_priority=true`, API will attempt to import editions
+    from ISBNs that don't exist in database immediately, rather
+    than queueing them for later lookup.
+    """
+
+    # Set up web context for dynlinks compatibility
+    # This ensures web.ctx.get("home") returns correct base URL
+    web.ctx.home = f"{request.url.scheme}://{request.url.netloc}"
+
+    # Build options dict from params (excluding bibkeys which goes to dynlinks separately)
+    options: DynlinksOptions = TypeAdapter(DynlinksOptions).validate_python(params.model_dump(exclude_unset=True))
+
+    # Format determination priority:
+    # 1. format parameter (highest)
+    # 2. callback parameter (medium) - when present, overrides path-based format
+    # 3. path-based detection (lowest)
+    if not params.format:
+        if params.callback:
+            options["format"] = "js"
+        elif request.url.path.endswith(".json"):
+            options["format"] = "json"
+
+    # Call existing business logic
+    result_str = dynlinks.dynlinks(bib_keys=params.bibkeys, options=options)
+
+    return wrap_jsonp(request, result_str)
+
+
+MULTIGET_PATH_RE = re.compile(r"/api/volumes/(brief|full)/json/(.+)")
+
+
+@router.get("/api/volumes/{brief_or_full}/json/{req}", include_in_schema=False)
+@router.get("/api/volumes/{brief_or_full}/json/{req}.json")
+async def get_volumes_multiget(
+    request: Request,
+    brief_or_full: Literal["brief", "full"],
+    req: str,
+) -> Any:
+    """
+    Get volume information for multiple identifiers.
+
+    This endpoint handles multi-lookup form of Hathi-style API,
+    allowing multiple identifiers in a single request.
+
+    Example:
+        GET /api/volumes/brief/json/isbn:059035342X|oclc:123456.json
+    """
+    web.ctx.home = f"{request.url.scheme}://{request.url.netloc}"
+
+    raw_uri = request.url.path
+
+    decoded_path = urllib.parse.unquote(raw_uri)
+
+    m = MULTIGET_PATH_RE.match(decoded_path)
+    if not m or len(m.groups()) != 2:
+        return {}
+
+    _brief_or_full, req = m.groups()
+
+    result = readlinks.readlinks(req, {})
+    return wrap_jsonp(request, result)
+
+
+# Left without a strict response model since it can return either a VolumeResponse dict or an empty list [] for non-existent identifiers. Pydantic
+# can't represent this union cleanly without losing validation benefits.
+@router.get("/api/volumes/{brief_or_full}/{idtype}/{idval}", include_in_schema=False)
+@router.get("/api/volumes/{brief_or_full}/{idtype}/{idval}.json")
+async def get_volume(
+    request: Request,
+    brief_or_full: Literal["brief", "full"],
+    idtype: Literal["oclc", "lccn", "issn", "isbn", "htid", "olid", "recordnumber"],
+    idval: str,
+    show_all_items: Annotated[bool, Query(description="Show all items including restricted ones")] = False,
+) -> Any:
+    """
+    Get volume information by identifier type and value.
+
+    This endpoint provides detailed information about a specific volume,
+    modeled after HathiTrust Bibliographic API. Includes information
+    about loans and other editions of same work.
+
+    Example:
+        GET /api/volumes/brief/isbn/059035342X.json
+    """
+    # Set up web context for readlinks compatibility
+    # This ensures web.ctx.get("home") returns correct base URL
+    web.ctx.home = f"{request.url.scheme}://{request.url.netloc}"
+
+    # Build request identifier
+    req = f"{idtype}:{idval}"
+
+    # Build options dict from query parameters
+    options: dict[str, str | bool] = {}
+    if show_all_items:
+        options["show_all_items"] = show_all_items
+
+    # Call existing business logic
+    result = readlinks.readlinks(req, options)
+
+    # Return the result for this specific request key
+    result = result.get(req, [])
+    return wrap_jsonp(request, result)

openlibrary/fastapi/models.py

@@ -1,13 +1,41 @@
 from __future__ import annotations
 
+import json
 import re
 from typing import Self
 
-from fastapi import HTTPException, Request
+from fastapi import HTTPException, Request, Response
 from pydantic import BaseModel, Field, model_validator
 
 from openlibrary.core.env import get_ol_env
 
+JS_CALLBACK_RE = re.compile(r"^[A-Za-z_$][A-Za-z0-9_$.]*$")
+
+
+def parse_comma_separated_list(v: str | list[str]) -> list[str]:
+    """
+    Parse comma-separated string values into a list of strings.
+
+    This validator handles both string and list inputs, converting:
+    - "a,b,c" → ["a", "b", "c"]
+    - ["a", "b,c"] → ["a", "b", "c"]
+
+    Used for query parameters that accept comma-separated values like:
+    - Search fields: "key,name,author_key"
+    - Bibliography keys: "ISBN1,ISBN2,ISBN3"
+
+    Args:
+        v: Input value (string or list of strings)
+
+    Returns:
+        List of trimmed strings with empty items filtered out
+    """
+    if not v:
+        return []
+    if isinstance(v, str):
+        v = [v]
+    return [f.strip() for item in v for f in str(item).split(",") if f.strip()]
+
 
 class Pagination(BaseModel):
     """Reusable pagination parameters for API endpoints."""
@@ -30,12 +58,18 @@ class PaginationLimit20(Pagination):
     limit: int = Field(20, ge=0, description="Maximum number of results to return.")
 
 
-def parse_comma_separated_list(v: str | list[str] | None) -> list[str]:
-    if not v:
-        return []
-    if isinstance(v, str):
-        v = [v]
-    return [f.strip() for item in v for f in str(item).split(",") if f.strip()]
+def wrap_jsonp(request: Request, data: dict | str) -> Response:
+    """Wrap data in JSONP callback if callback param is present.
+
+    Always returns a Response object.
+    Accepts either a dict (which will be JSON-serialized) or a pre-serialized JSON string.
+    """
+    json_string = json.dumps(data) if isinstance(data, dict) else data
+    if callback := request.query_params.get("callback"):
+        if not JS_CALLBACK_RE.match(callback):
+            raise ValueError("Invalid callback parameter: must be a valid JavaScript identifier (only letters, numbers, underscore, $, and . allowed)")
+        return Response(content=f"{callback}({json_string});", media_type="application/javascript")
+    return Response(content=json_string, media_type="application/json")
 
 
 class SolrInternalsParams(BaseModel):

openlibrary/plugins/books/code.py

@@ -5,12 +5,14 @@
 import urllib
 
 import web
+from typing_extensions import deprecated
 
 from infogami.plugins.api.code import jsonapi
 from infogami.utils import delegate
 from openlibrary.plugins.books import dynlinks, readlinks
 
 
+@deprecated("migrated to fastapi")
 class books_json(delegate.page):
     """
     Endpoint for mapping bib keys (e.g. ISBN, LCCN) to certain links associated
@@ -48,6 +50,7 @@ def GET(self):
         return dynlinks.dynlinks(bib_keys=i.bibkeys.split(","), options=i)
 
 
+@deprecated("migrated to fastapi")
 class read_singleget(delegate.page):
     """Handle the single-lookup form of the Hathi-style API"""
 
@@ -67,6 +70,7 @@ def GET(self, brief_or_full, idtype, idval):
         return json.dumps(result)
 
 
+@deprecated("migrated to fastapi")
 class read_multiget(delegate.page):
     """Handle the multi-lookup form of the Hathi-style API"""