fc3e: /api/posts blueprint — list (with cursor + filters) + detail

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/__init__.py

@@ -25,6 +25,7 @@ def all_blueprints() -> list[Blueprint]:
     from .import_admin import import_admin_bp
     from .ml_admin import ml_admin_bp
     from .platforms import platforms_bp
+    from .posts import posts_bp
     from .provenance import provenance_bp
     from .settings import settings_bp
     from .showcase import showcase_bp
@@ -48,6 +49,7 @@ def all_blueprints() -> list[Blueprint]:
         ml_admin_bp,
         sources_bp,
         platforms_bp,
+        posts_bp,
         credentials_bp,
         downloads_bp,
     ]

backend/app/api/posts.py

@@ -0,0 +1,69 @@
+"""FC-3e: /api/posts — cursor-paginated unified posts feed."""
+
+from quart import Blueprint, jsonify, request
+
+from ..extensions import get_session
+from ..services.post_feed_service import PostFeedService
+from ..services.source_service import KNOWN_PLATFORMS
+
+posts_bp = Blueprint("posts", __name__, url_prefix="/api/posts")
+
+
+def _bad(error: str, *, status: int = 400, detail: str | None = None, **extra):
+    body = {"error": error}
+    if detail is not None:
+        body["detail"] = detail
+    body.update(extra)
+    return jsonify(body), status
+
+
+@posts_bp.route("", methods=["GET"])
+async def list_posts():
+    args = request.args
+
+    cursor = args.get("cursor") or None
+    artist_id_raw = args.get("artist_id")
+    platform = args.get("platform") or None
+    limit_raw = args.get("limit", "24")
+
+    try:
+        limit = int(limit_raw)
+    except ValueError:
+        return _bad("invalid_limit", detail="limit must be an integer")
+    if limit < 1 or limit > 100:
+        return _bad("invalid_limit", detail="limit must be between 1 and 100")
+
+    artist_id = None
+    if artist_id_raw is not None:
+        try:
+            artist_id = int(artist_id_raw)
+        except ValueError:
+            return _bad("invalid_artist_id", detail="artist_id must be an integer")
+
+    if platform is not None and platform not in KNOWN_PLATFORMS:
+        return _bad(
+            "unknown_platform",
+            detail=f"platform must be one of {sorted(KNOWN_PLATFORMS)}",
+        )
+
+    async with get_session() as session:
+        try:
+            page = await PostFeedService(session).scroll(
+                cursor=cursor, artist_id=artist_id,
+                platform=platform, limit=limit,
+            )
+        except ValueError as exc:
+            # Service raises ValueError for malformed cursors only;
+            # limit bounds are validated above.
+            return _bad("invalid_cursor", detail=str(exc))
+
+    return jsonify(page)
+
+
+@posts_bp.route("/<int:post_id>", methods=["GET"])
+async def get_post(post_id: int):
+    async with get_session() as session:
+        item = await PostFeedService(session).get_post(post_id)
+    if item is None:
+        return _bad("not_found", status=404, detail=f"post id={post_id}")
+    return jsonify(item)

tests/test_api_posts.py

@@ -0,0 +1,119 @@
+"""FC-3e: /api/posts API tests.
+
+Validates list shape, cursor handling, filter validation, and detail
+endpoint. Service-level fixtures are exercised by test_post_feed_service
+— here we focus on the HTTP surface (validation, status codes, dict shape).
+"""
+from datetime import UTC, datetime
+
+import pytest
+
+from backend.app import create_app
+from backend.app.models import Artist, Post, Source
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def app():
+    return create_app()
+
+
+@pytest.fixture
+async def client(app):
+    async with app.test_client() as c:
+        yield c
+
+
+@pytest.fixture
+async def seeded_post(db):
+    artist = Artist(name="alice-api", slug="alice-api")
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://p/alice-api", enabled=True,
+    )
+    db.add(source)
+    await db.flush()
+    post = Post(
+        source_id=source.id, external_post_id="API1",
+        post_title="Hello", post_url="https://p/alice-api/1",
+        post_date=datetime.now(UTC),
+        description="<p>hi</p>",
+    )
+    db.add(post)
+    await db.commit()
+    return artist, source, post
+
+
+@pytest.mark.asyncio
+async def test_list_returns_items_and_cursor_keys(client, seeded_post):
+    resp = await client.get("/api/posts")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert set(body.keys()) == {"items", "next_cursor"}
+    assert isinstance(body["items"], list)
+    assert body["items"][0]["post_title"] == "Hello"
+    assert body["items"][0]["description_plain"] == "hi"
+
+
+@pytest.mark.asyncio
+async def test_list_rejects_malformed_cursor(client):
+    resp = await client.get("/api/posts?cursor=garbage!!!")
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "invalid_cursor"
+
+
+@pytest.mark.asyncio
+async def test_list_rejects_unknown_platform(client):
+    resp = await client.get("/api/posts?platform=myspace")
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "unknown_platform"
+
+
+@pytest.mark.asyncio
+async def test_list_rejects_non_int_artist_id(client):
+    resp = await client.get("/api/posts?artist_id=notanint")
+    assert resp.status_code == 400
+    body = await resp.get_json()
+    assert body["error"] == "invalid_artist_id"
+
+
+@pytest.mark.asyncio
+async def test_list_rejects_limit_out_of_range(client):
+    resp = await client.get("/api/posts?limit=0")
+    assert resp.status_code == 400
+    resp = await client.get("/api/posts?limit=500")
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_list_filter_propagates_artist(client, seeded_post):
+    artist, _, post = seeded_post
+    resp = await client.get(f"/api/posts?artist_id={artist.id}")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert len(body["items"]) == 1
+    assert body["items"][0]["id"] == post.id
+
+
+@pytest.mark.asyncio
+async def test_detail_200_for_known(client, seeded_post):
+    _, _, post = seeded_post
+    resp = await client.get(f"/api/posts/{post.id}")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["id"] == post.id
+    assert "description_full" in body
+    assert body["description_full"] == "hi"
+
+
+@pytest.mark.asyncio
+async def test_detail_404_for_unknown(client):
+    resp = await client.get("/api/posts/999999")
+    assert resp.status_code == 404
+    body = await resp.get_json()
+    assert body["error"] == "not_found"