migrate import_preview_json to FastAPI

openlibrary/asgi_app.py

@@ -215,6 +215,7 @@ def health() -> dict[str, str]:
     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.importapi import router as importapi_router
     from openlibrary.fastapi.internal.api import router as internal_router
     from openlibrary.fastapi.languages import router as languages_router
     from openlibrary.fastapi.lists import router as lists_router
@@ -232,6 +233,7 @@ def health() -> dict[str, str]:
     app.include_router(books_router)
     app.include_router(cdn_router)
     app.include_router(checkins_router)
+    app.include_router(importapi_router)
     app.include_router(internal_router)
     app.include_router(languages_router)
     app.include_router(lists_router)

openlibrary/fastapi/importapi.py

@@ -0,0 +1,112 @@
+"""
+FastAPI endpoints for import preview.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+import web
+from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, status
+
+from openlibrary.accounts import get_current_user
+from openlibrary.fastapi.auth import AuthenticatedUser, require_authenticated_user
+from openlibrary.plugins.importapi.import_ui import ImportPreviewRequest
+from openlibrary.utils.request_context import site, web_ctx_ip
+
+router = APIRouter(tags=["import"])
+
+
+def _client_ip(request: Request) -> str:
+    if forwarded := request.headers.get("X-Forwarded-For"):
+        return forwarded.split(",")[0].strip()
+    if real_ip := request.headers.get("X-Real-IP"):
+        return real_ip.strip()
+    if request.client:
+        return request.client.host
+    return "127.0.0.1"
+
+
+def _build_preview_response(
+    source: str | None,
+    provider: str | None,
+    identifier: str | None,
+    save: bool,
+    request: Request,
+) -> dict:
+    web.ctx.site = site.get()
+
+    try:
+        req = ImportPreviewRequest.from_input(
+            {
+                "source": source,
+                "provider": provider,
+                "identifier": identifier,
+                "save": "true" if save else "false",
+            }
+        )
+    except ValueError as e:
+        return {"success": False, "error": str(e)}
+
+    with web_ctx_ip(_client_ip(request)):
+        return dict(req.metadata_provider.do_import(req.identifier, req.save))
+
+
+@router.get("/import/preview.json")
+def import_preview_json_get(
+    request: Request,
+    _: Annotated[AuthenticatedUser, Depends(require_authenticated_user)],
+    source: Annotated[
+        str | None,
+        Query(
+            description="Source in format 'provider:identifier' (e.g. 'amazon:ASIN')",
+        ),
+    ] = None,
+    provider: Annotated[
+        str | None,
+        Query(
+            description="Metadata provider: amazon, ia, or marc",
+        ),
+    ] = None,
+    identifier: Annotated[
+        str | None,
+        Query(
+            description="Identifier for the given provider",
+        ),
+    ] = None,
+) -> dict:
+    """
+    Preview import metadata without saving.
+
+    Requires admin, librarian, or super-librarian role.
+    """
+    user = get_current_user()
+    if not (user and (user.is_admin() or user.is_librarian() or user.is_super_librarian())):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Insufficient permissions",
+        )
+    return _build_preview_response(source, provider, identifier, save=False, request=request)
+
+
+@router.post("/import/preview.json")
+def import_preview_json_post(
+    request: Request,
+    _: Annotated[AuthenticatedUser, Depends(require_authenticated_user)],
+    source: Annotated[str | None, Form()] = None,
+    provider: Annotated[str | None, Form()] = None,
+    identifier: Annotated[str | None, Form()] = None,
+    save: Annotated[bool, Form()] = False,
+) -> dict:
+    """
+    Import metadata, optionally saving it.
+
+    Requires admin, librarian, or super-librarian role.
+    """
+    user = get_current_user()
+    if not (user and (user.is_admin() or user.is_librarian() or user.is_super_librarian())):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Insufficient permissions",
+        )
+    return _build_preview_response(source, provider, identifier, save=save, request=request)

openlibrary/plugins/importapi/import_ui.py

@@ -1,6 +1,7 @@
 import json
 from dataclasses import dataclass
 from typing import Literal, cast, override
+from warnings import deprecated
 
 import requests
 import web
@@ -62,6 +63,7 @@ def POST(self):
             )
 
 
+@deprecated("migrated to fastapi")
 class import_preview_json(delegate.page):
     path = "/import/preview"
     encoding = "json"

openlibrary/tests/fastapi/test_importapi.py

@@ -0,0 +1,253 @@
+"""Tests for the FastAPI import preview endpoints."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _raise(exc: Exception) -> None:
+    raise exc
+
+
+FAKE_IMPORT_RESULT = {
+    "edition": {"key": "/books/OL1M", "title": "Test Book"},
+    "success": True,
+}
+
+
+@pytest.fixture
+def mock_site():
+    """Mock the site ContextVar to avoid real infogami connection."""
+    with patch("openlibrary.fastapi.importapi.site") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_user_factory(mock_site, monkeypatch):
+    """Factory fixture to create and mock users with configurable roles.
+
+    Usage:
+        user = mock_user_factory(is_admin=True)
+        user = mock_user_factory(is_librarian=True)
+    """
+
+    def create_user(
+        is_admin: bool = False,
+        is_librarian: bool = False,
+        is_super_librarian: bool = False,
+    ):
+        user = MagicMock()
+        user.is_admin.return_value = is_admin
+        user.is_librarian.return_value = is_librarian
+        user.is_super_librarian.return_value = is_super_librarian
+        monkeypatch.setattr("openlibrary.fastapi.importapi.get_current_user", lambda: user)
+        return user
+
+    return create_user
+
+
+@pytest.fixture
+def mock_import_request():
+    """Return a factory that creates a mock ImportPreviewRequest."""
+
+    def _make(save: bool = False):
+        mock_req = MagicMock()
+        mock_req.metadata_provider.do_import.return_value = FAKE_IMPORT_RESULT
+        mock_req.save = save
+        return mock_req
+
+    return _make
+
+
+class TestImportPreviewAuth:
+    """Authentication and authorization tests."""
+
+    def test_get_requires_authentication(self, fastapi_client, mock_site):
+        response = fastapi_client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 401
+
+    def test_post_requires_authentication(self, fastapi_client, mock_site):
+        response = fastapi_client.post("/import/preview.json", data={"source": "amazon:ASIN123"})
+        assert response.status_code == 401
+
+    def test_get_forbidden_for_regular_user(self, fastapi_client, mock_authenticated_user, mock_user_factory):
+        mock_user_factory()
+        response = fastapi_client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 403
+
+    def test_post_forbidden_for_regular_user(self, fastapi_client, mock_authenticated_user, mock_user_factory):
+        mock_user_factory()
+        response = fastapi_client.post("/import/preview.json", data={"source": "amazon:ASIN123"})
+        assert response.status_code == 403
+
+    def test_get_allows_admin(self, fastapi_client, mock_authenticated_user, mock_user_factory, mock_import_request, monkeypatch):
+        mock_user_factory(is_admin=True)
+        monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: mock_import_request(),
+        )
+        response = fastapi_client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 200
+
+    def test_get_allows_librarian(self, fastapi_client, mock_authenticated_user, mock_user_factory, mock_import_request, monkeypatch):
+        mock_user_factory(is_librarian=True)
+        monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: mock_import_request(),
+        )
+        response = fastapi_client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 200
+
+    def test_get_allows_super_librarian(self, fastapi_client, mock_authenticated_user, mock_user_factory, mock_import_request, monkeypatch):
+        mock_user_factory(is_super_librarian=True)
+        monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: mock_import_request(),
+        )
+        response = fastapi_client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 200
+
+
+class TestImportPreviewGet:
+    """GET /import/preview.json endpoint tests."""
+
+    @pytest.fixture(autouse=True)
+    def _auth_setup(self, fastapi_client, mock_authenticated_user, mock_user_factory, monkeypatch):
+        self.client = fastapi_client
+        self.monkeypatch = monkeypatch
+        mock_user_factory(is_admin=True)
+
+    def test_get_returns_import_result(self, mock_import_request):
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: mock_import_request(),
+        )
+        response = self.client.get("/import/preview.json?source=amazon:ASIN123")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is True
+        assert data["edition"]["key"] == "/books/OL1M"
+
+    def test_get_does_not_save(self):
+        captured = {}
+
+        def fake_from_input(i):
+            captured["save"] = i.get("save", "false")
+            mock_req = MagicMock()
+            mock_req.metadata_provider.do_import.return_value = FAKE_IMPORT_RESULT
+            mock_req.save = False
+            return mock_req
+
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            fake_from_input,
+        )
+        response = self.client.get("/import/preview.json?source=amazon:ASIN123&save=true")
+        assert response.status_code == 200
+        assert captured["save"] == "false"
+
+    def test_get_invalid_source_returns_error(self):
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: _raise(ValueError("Invalid source provided")),
+        )
+        response = self.client.get("/import/preview.json?source=invalid")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is False
+        assert "Invalid source provided" in data["error"]
+
+    def test_get_with_provider_and_identifier(self):
+        captured = {}
+
+        def fake_from_input(i):
+            captured["params"] = i
+            mock_req = MagicMock()
+            mock_req.metadata_provider.do_import.return_value = FAKE_IMPORT_RESULT
+            mock_req.save = False
+            return mock_req
+
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            fake_from_input,
+        )
+        response = self.client.get("/import/preview.json?provider=amazon&identifier=1234567890")
+        assert response.status_code == 200
+        assert captured["params"]["provider"] == "amazon"
+        assert captured["params"]["identifier"] == "1234567890"
+
+    def test_get_no_params_returns_error(self):
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: _raise(ValueError("No provider specified")),
+        )
+        response = self.client.get("/import/preview.json")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is False
+
+
+class TestImportPreviewPost:
+    """POST /import/preview.json endpoint tests."""
+
+    @pytest.fixture(autouse=True)
+    def _auth_setup(self, fastapi_client, mock_authenticated_user, mock_user_factory, monkeypatch):
+        self.client = fastapi_client
+        self.monkeypatch = monkeypatch
+        mock_user_factory(is_admin=True)
+
+    def test_post_returns_import_result(self, mock_import_request):
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: mock_import_request(),
+        )
+        response = self.client.post("/import/preview.json", data={"source": "amazon:ASIN123"})
+        assert response.status_code == 200
+
+    def test_post_can_save(self):
+        captured = {}
+
+        def fake_from_input(i):
+            captured["save"] = i.get("save", "false")
+            mock_req = MagicMock()
+            mock_req.metadata_provider.do_import.return_value = FAKE_IMPORT_RESULT
+            mock_req.save = captured["save"] == "true"
+            return mock_req
+
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            fake_from_input,
+        )
+        response = self.client.post(
+            "/import/preview.json",
+            data={"source": "amazon:ASIN123", "save": "true"},
+        )
+        assert response.status_code == 200
+        assert captured["save"] == "true"
+
+    def test_post_without_save_defaults_to_false(self):
+        captured = {}
+
+        def fake_from_input(i):
+            captured["save"] = i.get("save", "false")
+            mock_req = MagicMock()
+            mock_req.metadata_provider.do_import.return_value = FAKE_IMPORT_RESULT
+            return mock_req
+
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            fake_from_input,
+        )
+        response = self.client.post("/import/preview.json", data={"source": "amazon:ASIN123"})
+        assert response.status_code == 200
+        assert captured["save"] == "false"
+
+    def test_post_invalid_source_returns_error(self):
+        self.monkeypatch.setattr(
+            "openlibrary.fastapi.importapi.ImportPreviewRequest.from_input",
+            lambda i: _raise(ValueError("Invalid source provided")),
+        )
+        response = self.client.post("/import/preview.json", data={"source": "invalid"})
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is False