feat(provenance): _artists_for resolves via artist_id (closes vii-b gap)

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

backend/app/services/gallery_service.py

@@ -13,7 +13,7 @@ from datetime import datetime
 from sqlalchemy import and_, exists, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag
+from ..models import Artist, ImageProvenance, ImageRecord, Source, Tag
 from ..models.tag import image_tag
 
 CURSOR_SEPARATOR = "|"
@@ -94,44 +94,19 @@ def _provenance_clause(post_id, artist_id):
 
 
 async def _artists_for(session, image_ids: list[int]) -> dict[int, dict]:
-    """Map image_id -> {"name","slug"} using the primary-post path first,
+    """Map image_id -> {"name","slug"} via the canonical
-    then the lowest-id ImageProvenance row as fallback. Bounded by page
+    image_record.artist_id (FC-2d-vii-c). Bounded by page size."""
-    size (<= 200) → two indexed selects, no N+1."""
     if not image_ids:
         return {}
-    out: dict[int, dict] = {}
+    stmt = (
-
-    primary = (
         select(ImageRecord.id, Artist.name, Artist.slug)
-        .join(Post, Post.id == ImageRecord.primary_post_id)
+        .join(Artist, Artist.id == ImageRecord.artist_id)
-        .join(Source, Source.id == Post.source_id)
-        .join(Artist, Artist.id == Source.artist_id)
         .where(ImageRecord.id.in_(image_ids))
     )
-    for img_id, name, slug in (await session.execute(primary)).all():
+    return {
-        out[img_id] = {"name": name, "slug": slug}
+        img_id: {"name": name, "slug": slug}
-
+        for img_id, name, slug in (await session.execute(stmt)).all()
-    remaining = [i for i in image_ids if i not in out]
+    }
-    if remaining:
-        fb = (
-            select(
-                ImageProvenance.image_record_id,
-                ImageProvenance.id,
-                Artist.name,
-                Artist.slug,
-            )
-            .join(Source, Source.id == ImageProvenance.source_id)
-            .join(Artist, Artist.id == Source.artist_id)
-            .where(ImageProvenance.image_record_id.in_(remaining))
-            .order_by(
-                ImageProvenance.image_record_id.asc(),
-                ImageProvenance.id.asc(),
-            )
-        )
-        for img_id, _pid, name, slug in (await session.execute(fb)).all():
-            if img_id not in out:  # first (lowest id) wins
-                out[img_id] = {"name": name, "slug": slug}
-    return out
 
 
 class GalleryService:

tests/test_gallery_artist_payload.py

@@ -3,13 +3,7 @@ from datetime import UTC, datetime, timedelta
 import pytest
 
 from backend.app import create_app
-from backend.app.models import (
+from backend.app.models import Artist, ImageRecord, Source
-    Artist,
-    ImageProvenance,
-    ImageRecord,
-    Post,
-    Source,
-)
 
 pytestmark = pytest.mark.integration
 
@@ -48,42 +42,6 @@ async def _artist_source(db, name, slug):
     return a, s
 
 
-@pytest.mark.asyncio
-async def test_scroll_artist_via_primary_post(client, db):
-    rec = await _img(db, 1)
-    _, s = await _artist_source(db, "Alice", "alice")
-    post = Post(source_id=s.id, external_post_id="1")
-    db.add(post)
-    await db.flush()
-    rec.primary_post_id = post.id
-    await db.flush()
-    await db.commit()
-
-    resp = await client.get("/api/gallery/scroll?limit=10")
-    assert resp.status_code == 200
-    body = await resp.get_json()
-    item = next(i for i in body["images"] if i["id"] == rec.id)
-    assert item["artist"] == {"name": "Alice", "slug": "alice"}
-
-
-@pytest.mark.asyncio
-async def test_scroll_artist_via_provenance_fallback(client, db):
-    rec = await _img(db, 1)
-    _, s = await _artist_source(db, "Bob", "bob")
-    post = Post(source_id=s.id, external_post_id="2")
-    db.add(post)
-    await db.flush()
-    db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
-                           source_id=s.id))
-    await db.flush()
-    await db.commit()  # rec.primary_post_id stays NULL
-
-    resp = await client.get("/api/gallery/scroll?limit=10")
-    body = await resp.get_json()
-    item = next(i for i in body["images"] if i["id"] == rec.id)
-    assert item["artist"] == {"name": "Bob", "slug": "bob"}
-
-
 @pytest.mark.asyncio
 async def test_scroll_artist_null_when_none(client, db):
     rec = await _img(db, 1)
@@ -92,3 +50,20 @@ async def test_scroll_artist_null_when_none(client, db):
     body = await resp.get_json()
     item = next(i for i in body["images"] if i["id"] == rec.id)
     assert item["artist"] is None
+
+
+@pytest.mark.asyncio
+async def test_scroll_artist_via_artist_id(client, db):
+    # FC-2d-vii-c: the canonical resolution path is image_record.artist_id
+    # (covers folder-style images with no provenance too).
+    rec = await _img(db, 1)
+    a, _ = await _artist_source(db, "Folder", "folder")
+    rec.artist_id = a.id
+    await db.flush()
+    await db.commit()
+
+    resp = await client.get("/api/gallery/scroll?limit=10")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    item = next(i for i in body["images"] if i["id"] == rec.id)
+    assert item["artist"] == {"name": "Folder", "slug": "folder"}