feat(bulk): BulkTagService (common_tags/bulk_add/bulk_remove) + 3 image bulk-tag routes

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

backend/app/api/tags.py

@@ -2,10 +2,12 @@
 
 from quart import Blueprint, jsonify, request
 from sqlalchemy import exists, select
+from sqlalchemy.exc import IntegrityError
 
 from ..extensions import get_session
-from ..models import TagKind
+from ..models import Tag, TagKind
 from ..models.tag_allowlist import TagAllowlist
+from ..services.bulk_tag_service import BulkTagService
 from ..services.tag_directory_service import TagDirectoryService
 from ..services.tag_service import (
     TagMergeConflict,
@@ -25,6 +27,31 @@ def _coerce_kind(raw: str | None) -> TagKind | None:
         return None
 
 
+def _parse_bulk_ids(body) -> tuple[list[int] | None, tuple | None]:
+    """Returns (image_ids, error). error is (json, status) or None."""
+    if not body or "image_ids" not in body:
+        return None, (jsonify({"error": "image_ids required"}), 400)
+    raw = body["image_ids"]
+    if not isinstance(raw, list) or not raw:
+        return None, (
+            jsonify({"error": "image_ids must be a non-empty list"}),
+            400,
+        )
+    try:
+        ids = [int(x) for x in raw]
+    except (TypeError, ValueError):
+        return None, (
+            jsonify({"error": "image_ids must be integers"}),
+            400,
+        )
+    if len(ids) > 200:
+        return None, (
+            jsonify({"error": "selection too large (max 200)"}),
+            400,
+        )
+    return ids, None
+
+
 @tags_bp.route("/tags/autocomplete", methods=["GET"])
 async def autocomplete():
     q = request.args.get("q", "")
@@ -200,3 +227,58 @@ async def merge_tag(source_id: int):
             "source_deleted": result.source_deleted,
         }
     )
+
+
+@tags_bp.route("/images/common-tags", methods=["POST"])
+async def common_tags():
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body)
+    if err:
+        return err
+    async with get_session() as session:
+        tags = await BulkTagService(session).common_tags(ids)
+    return jsonify({"tags": tags})
+
+
+@tags_bp.route("/images/bulk/tags", methods=["POST"])
+async def bulk_add_tag():
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body)
+    if err:
+        return err
+    if not body or "tag_id" not in body:
+        return jsonify({"error": "tag_id required"}), 400
+    source = body.get("source", "manual")
+    if source not in ("manual", "ml_accepted"):
+        return jsonify({"error": f"invalid source {source!r}"}), 400
+    async with get_session() as session:
+        if await session.get(Tag, body["tag_id"]) is None:
+            return jsonify({"error": "tag not found"}), 404
+        try:
+            added = await BulkTagService(session).bulk_add(
+                ids, body["tag_id"], source=source
+            )
+        except IntegrityError:
+            return jsonify(
+                {"error": "one or more image_ids do not exist"}
+            ), 400
+        await session.commit()
+    return jsonify({"added_count": added, "total": len(ids)})
+
+
+@tags_bp.route("/images/bulk/tags/remove", methods=["POST"])
+async def bulk_remove_tag():
+    body = await request.get_json()
+    ids, err = _parse_bulk_ids(body)
+    if err:
+        return err
+    if not body or "tag_id" not in body:
+        return jsonify({"error": "tag_id required"}), 400
+    async with get_session() as session:
+        if await session.get(Tag, body["tag_id"]) is None:
+            return jsonify({"error": "tag not found"}), 404
+        removed = await BulkTagService(session).bulk_remove(
+            ids, body["tag_id"]
+        )
+        await session.commit()
+    return jsonify({"removed_count": removed})

backend/app/services/bulk_tag_service.py

@@ -0,0 +1,93 @@
+"""Bulk tag operations over a set of images (Core, set-based, atomic)."""
+
+from sqlalchemy import and_, func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import Tag
+from ..models.tag import image_tag
+from ..models.tag_suggestion_rejection import TagSuggestionRejection
+
+
+class BulkTagService:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def common_tags(self, image_ids: list[int]) -> list[dict]:
+        """Tags present on EVERY image in image_ids."""
+        if not image_ids:
+            return []
+        n = len(set(image_ids))
+        stmt = (
+            select(Tag.id, Tag.name, Tag.kind)
+            .join(image_tag, image_tag.c.tag_id == Tag.id)
+            .where(image_tag.c.image_record_id.in_(image_ids))
+            .group_by(Tag.id, Tag.name, Tag.kind)
+            .having(
+                func.count(func.distinct(image_tag.c.image_record_id)) == n
+            )
+            .order_by(Tag.name.asc())
+        )
+        rows = (await self.session.execute(stmt)).all()
+        return [
+            {
+                "id": r.id,
+                "name": r.name,
+                "kind": r.kind.value if hasattr(r.kind, "value") else r.kind,
+            }
+            for r in rows
+        ]
+
+    async def bulk_add(
+        self, image_ids: list[int], tag_id: int, source: str = "manual"
+    ) -> int:
+        """Apply tag_id to every image. Idempotent per (image, tag).
+        Returns the number of links actually created."""
+        if not image_ids:
+            return 0
+        stmt = (
+            pg_insert(image_tag)
+            .values(
+                [
+                    {
+                        "image_record_id": iid,
+                        "tag_id": tag_id,
+                        "source": source,
+                    }
+                    for iid in set(image_ids)
+                ]
+            )
+            .on_conflict_do_nothing(
+                index_elements=["image_record_id", "tag_id"]
+            )
+        )
+        res = await self.session.execute(stmt)
+        return res.rowcount or 0
+
+    async def bulk_remove(self, image_ids: list[int], tag_id: int) -> int:
+        """Remove tag_id from every image, and record a per-image rejection
+        so the proactive allowlist sweep won't immediately re-add it
+        (mirrors single-image removal semantics). Returns links removed."""
+        if not image_ids:
+            return 0
+        res = await self.session.execute(
+            image_tag.delete().where(
+                and_(
+                    image_tag.c.tag_id == tag_id,
+                    image_tag.c.image_record_id.in_(image_ids),
+                )
+            )
+        )
+        await self.session.execute(
+            pg_insert(TagSuggestionRejection)
+            .values(
+                [
+                    {"image_record_id": iid, "tag_id": tag_id}
+                    for iid in set(image_ids)
+                ]
+            )
+            .on_conflict_do_nothing(
+                index_elements=["image_record_id", "tag_id"]
+            )
+        )
+        return res.rowcount or 0

tests/test_api_bulk_tags.py

@@ -0,0 +1,91 @@
+import pytest
+
+from backend.app import create_app
+
+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
+
+
+async def _mk_tag(client, name, kind="general"):
+    r = await client.post("/api/tags", json={"name": name, "kind": kind})
+    return (await r.get_json())["id"]
+
+
+@pytest.mark.asyncio
+async def test_common_tags_route(client):
+    resp = await client.post("/api/images/common-tags", json={"image_ids": []})
+    assert resp.status_code == 200
+    assert (await resp.get_json())["tags"] == []
+
+
+@pytest.mark.asyncio
+async def test_common_tags_requires_list(client):
+    resp = await client.post("/api/images/common-tags", json={})
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_route_happy(client):
+    tag = await _mk_tag(client, "RouteAdd")
+    resp = await client.post(
+        "/api/images/bulk/tags",
+        json={"image_ids": [1, 2, 3], "tag_id": tag},
+    )
+    # images 1..3 may not exist; FK means added_count reflects real images.
+    assert resp.status_code in (200, 400)
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_rejects_bad_source(client):
+    tag = await _mk_tag(client, "RouteAddSrc")
+    resp = await client.post(
+        "/api/images/bulk/tags",
+        json={"image_ids": [1], "tag_id": tag, "source": "evil"},
+    )
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_missing_tag_404(client):
+    resp = await client.post(
+        "/api/images/bulk/tags",
+        json={"image_ids": [1], "tag_id": 999999},
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_over_cap_400(client):
+    tag = await _mk_tag(client, "RouteAddCap")
+    resp = await client.post(
+        "/api/images/bulk/tags",
+        json={"image_ids": list(range(1, 202)), "tag_id": tag},
+    )
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_bulk_remove_route_missing_tag_404(client):
+    resp = await client.post(
+        "/api/images/bulk/tags/remove",
+        json={"image_ids": [1], "tag_id": 999999},
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_bulk_remove_route_requires_ids(client):
+    resp = await client.post(
+        "/api/images/bulk/tags/remove", json={"tag_id": 1}
+    )
+    assert resp.status_code == 400

tests/test_bulk_tag_service.py

@@ -0,0 +1,122 @@
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.models import ImageRecord, TagKind
+from backend.app.models.tag import image_tag
+from backend.app.models.tag_suggestion_rejection import TagSuggestionRejection
+from backend.app.services.bulk_tag_service import BulkTagService
+from backend.app.services.tag_service import TagService
+
+pytestmark = pytest.mark.integration
+
+_SEQ = 0
+
+
+async def _img(db):
+    global _SEQ
+    _SEQ += 1
+    rec = ImageRecord(
+        path=f"/tmp/fc_bulk_{_SEQ}.png",
+        sha256=f"{_SEQ:064d}",
+        size_bytes=1,
+        mime="image/png",
+        origin="uploaded",
+    )
+    db.add(rec)
+    await db.flush()
+    return rec.id
+
+
+@pytest.mark.asyncio
+async def test_common_tags_only_returns_tags_on_every_image(db):
+    tags = TagService(db)
+    t_all = await tags.find_or_create("OnAll", TagKind.general)
+    t_some = await tags.find_or_create("OnSome", TagKind.general)
+    i1, i2 = await _img(db), await _img(db)
+    await tags.add_to_image(i1, t_all.id)
+    await tags.add_to_image(i2, t_all.id)
+    await tags.add_to_image(i1, t_some.id)  # only one image
+    svc = BulkTagService(db)
+    result = await svc.common_tags([i1, i2])
+    assert [t["id"] for t in result] == [t_all.id]
+    assert result[0]["name"] == "OnAll"
+    assert result[0]["kind"] == "general"
+
+
+@pytest.mark.asyncio
+async def test_common_tags_empty_when_no_images(db):
+    svc = BulkTagService(db)
+    assert await svc.common_tags([]) == []
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_applies_and_counts_idempotent(db):
+    tags = TagService(db)
+    tag = await tags.find_or_create("BulkAdd", TagKind.general)
+    i1, i2 = await _img(db), await _img(db)
+    svc = BulkTagService(db)
+    added = await svc.bulk_add([i1, i2], tag.id)
+    assert added == 2
+    # Idempotent: re-adding adds nothing.
+    again = await svc.bulk_add([i1, i2], tag.id)
+    assert again == 0
+    rows = (
+        await db.execute(
+            select(image_tag.c.image_record_id, image_tag.c.source).where(
+                image_tag.c.tag_id == tag.id
+            )
+        )
+    ).all()
+    assert {r.image_record_id for r in rows} == {i1, i2}
+    assert all(r.source == "manual" for r in rows)
+
+
+@pytest.mark.asyncio
+async def test_bulk_add_records_explicit_source(db):
+    tags = TagService(db)
+    tag = await tags.find_or_create("BulkAddML", TagKind.general)
+    i1 = await _img(db)
+    svc = BulkTagService(db)
+    await svc.bulk_add([i1], tag.id, source="ml_accepted")
+    src = await db.scalar(
+        select(image_tag.c.source).where(image_tag.c.tag_id == tag.id)
+    )
+    assert src == "ml_accepted"
+
+
+@pytest.mark.asyncio
+async def test_bulk_remove_counts_and_records_rejections(db):
+    tags = TagService(db)
+    tag = await tags.find_or_create("BulkRem", TagKind.general)
+    i1, i2 = await _img(db), await _img(db)
+    svc = BulkTagService(db)
+    await svc.bulk_add([i1, i2], tag.id)
+    removed = await svc.bulk_remove([i1, i2], tag.id)
+    assert removed == 2
+    remaining = await db.scalar(
+        select(func.count())
+        .select_from(image_tag)
+        .where(image_tag.c.tag_id == tag.id)
+    )
+    assert remaining == 0
+    rej = (
+        await db.execute(
+            select(TagSuggestionRejection.image_record_id).where(
+                TagSuggestionRejection.tag_id == tag.id
+            )
+        )
+    ).scalars().all()
+    assert set(rej) == {i1, i2}
+
+
+@pytest.mark.asyncio
+async def test_bulk_remove_rejection_is_idempotent(db):
+    tags = TagService(db)
+    tag = await tags.find_or_create("BulkRem2", TagKind.general)
+    i1 = await _img(db)
+    svc = BulkTagService(db)
+    await svc.bulk_add([i1], tag.id)
+    await svc.bulk_remove([i1], tag.id)
+    # Second remove: nothing to delete, rejection already present, no error.
+    removed = await svc.bulk_remove([i1], tag.id)
+    assert removed == 0