feat(tags): edit a character tag's fandom (backend)
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 23s
CI / backend-lint-and-test (push) Successful in 27s
CI / intimp (push) Successful in 3m41s
CI / intapi (push) Successful in 7m46s
CI / intcore (push) Successful in 8m29s

No way existed to change which fandom a character tag belongs to after
creation — PATCH /tags/<id> only renamed.

- TagService.set_fandom(tag_id, fandom_id, merge=False): set / change /
  clear (fandom_id=None) a character's fandom, with the same validation as
  find_or_create. On a name collision in the target fandom it raises
  TagMergeConflict (→ 409, same shape as rename); merge=True resolves it by
  merging this tag INTO the existing character.
- Extract _do_merge(source, target) from merge() so set_fandom can perform
  the deliberate CROSS-fandom merge the public merge() validation forbids.
- PATCH /tags/<id> now accepts optional fandom_id (+ merge flag) alongside
  name, and returns fandom_id.

Tests: set/change/clear, non-character + bad-ref rejection, collision
raises, merge resolves; API set/clear + collision→merge.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-03 23:13:24 -04:00
parent e05e0b9f37
commit d9ab6e15c6
4 changed files with 195 additions and 9 deletions
+24 -6
View File
@@ -195,14 +195,27 @@ async def remove_tag_from_image(image_id: int, tag_id: int):
@tags_bp.route("/tags/<int:tag_id>", methods=["PATCH"])
async def rename_tag(tag_id: int):
body = await request.get_json()
if not body or "name" not in body:
return jsonify({"error": "name required"}), 400
async def update_tag(tag_id: int):
"""Rename and/or re-fandom a tag. Body may carry `name` and/or
`fandom_id` (a fandom tag id, or null to clear — character tags only).
`merge: true` resolves a collision by merging into the existing tag.
"""
body = await request.get_json() or {}
has_name = "name" in body
has_fandom = "fandom_id" in body
if not has_name and not has_fandom:
return jsonify({"error": "name or fandom_id required"}), 400
do_merge = bool(body.get("merge"))
async with get_session() as session:
svc = TagService(session)
try:
tag = await svc.rename(tag_id, body["name"])
tag = None
if has_name:
tag = await svc.rename(tag_id, body["name"])
if has_fandom:
tag = await svc.set_fandom(
tag_id, body["fandom_id"], merge=do_merge
)
except TagMergeConflict as exc:
return jsonify(
{
@@ -219,7 +232,12 @@ async def rename_tag(tag_id: int):
return jsonify({"error": str(exc)}), 400
await session.commit()
return jsonify(
{"id": tag.id, "name": tag.name, "kind": tag.kind.value}
{
"id": tag.id,
"name": tag.name,
"kind": tag.kind.value,
"fandom_id": tag.fandom_id,
}
)
+70
View File
@@ -280,6 +280,69 @@ class TagService:
await self.session.flush()
return tag
async def set_fandom(
self, tag_id: int, fandom_id: int | None, *, merge: bool = False
) -> Tag:
"""Set / change / clear a character tag's fandom.
Raises TagValidationError unless the tag is a character and fandom_id
(when given) references a fandom tag. If the change would collide with
an existing character of the same name in the TARGET fandom, raises
TagMergeConflict (the API turns that into a 409 merge hint) — unless
merge=True, in which case this tag is merged INTO that existing
character (a deliberate cross-fandom merge) and the surviving target
is returned. Passing fandom_id=None clears the fandom.
"""
tag = await self.session.get(Tag, tag_id)
if tag is None:
raise TagValidationError(f"Tag {tag_id} not found")
if tag.kind != TagKind.character:
raise TagValidationError("Only character tags can have a fandom")
if fandom_id is not None:
fandom = await self.session.get(Tag, fandom_id)
if fandom is None or fandom.kind != TagKind.fandom:
raise TagValidationError(
f"fandom_id {fandom_id} does not reference a fandom tag"
)
if fandom_id == tag.fandom_id:
return tag
# Collision: another character with the same name already lives in the
# target fandom. Mirrors rename's (name, kind, fandom_id) uniqueness.
clash_stmt = (
select(Tag)
.where(Tag.name == tag.name)
.where(Tag.kind == TagKind.character)
.where(
Tag.fandom_id.is_(None)
if fandom_id is None
else Tag.fandom_id == fandom_id
)
.where(Tag.id != tag_id)
)
clash = (await self.session.execute(clash_stmt)).scalar_one_or_none()
if clash is not None:
if not merge:
source_image_count = await self.session.scalar(
select(func.count())
.select_from(image_tag)
.where(image_tag.c.tag_id == tag_id)
)
will_alias = await self._keep_as_alias(tag_id)
raise TagMergeConflict(
f"A character named {tag.name!r} already exists in that fandom",
target_id=clash.id,
target_name=clash.name,
source_image_count=int(source_image_count or 0),
will_alias=will_alias,
)
await self._do_merge(tag, clash)
return clash
tag.fandom_id = fandom_id
await self.session.flush()
return tag
async def merge(self, source_id: int, target_id: int) -> MergeResult:
"""Transactionally repoint every FK from source→target, optionally
keep source's name as a tagger alias, delete source. Atomic: any
@@ -298,7 +361,14 @@ class TagService:
raise TagValidationError(
"Tags must be the same kind and fandom to merge"
)
return await self._do_merge(source, target)
async def _do_merge(self, source: Tag, target: Tag) -> MergeResult:
"""Repoint every FK source→target, optionally keep source's name as a
tagger alias, delete source. NO kind/fandom validation — callers that
need it (public merge()) validate first; set_fandom's collision
resolution calls this directly for a deliberate CROSS-fandom merge."""
source_id, target_id = source.id, target.id
keep_as_alias = await self._keep_as_alias(source_id)
source_name = source.name
source_kind = source.kind
+35 -2
View File
@@ -3,11 +3,44 @@ import pytest
pytestmark = pytest.mark.integration
async def _mk(client, name, kind):
r = await client.post("/api/tags", json={"name": name, "kind": kind})
async def _mk(client, name, kind, fandom_id=None):
body = {"name": name, "kind": kind}
if fandom_id is not None:
body["fandom_id"] = fandom_id
r = await client.post("/api/tags", json=body)
return (await r.get_json())["id"]
@pytest.mark.asyncio
async def test_patch_sets_and_clears_character_fandom(client):
fandom = await _mk(client, "Bleach", "fandom")
char = await _mk(client, "Ichigo", "character")
resp = await client.patch(f"/api/tags/{char}", json={"fandom_id": fandom})
assert resp.status_code == 200
assert (await resp.get_json())["fandom_id"] == fandom
resp = await client.patch(f"/api/tags/{char}", json={"fandom_id": None})
assert resp.status_code == 200
assert (await resp.get_json())["fandom_id"] is None
@pytest.mark.asyncio
async def test_patch_fandom_collision_then_merge(client):
f1 = await _mk(client, "Bleach", "fandom")
f2 = await _mk(client, "Naruto", "fandom")
c1 = await _mk(client, "Renji", "character", fandom_id=f1)
c2 = await _mk(client, "Renji", "character", fandom_id=f2)
# Moving c1 into f2 collides with c2 → rich 409 (same shape as rename).
resp = await client.patch(f"/api/tags/{c1}", json={"fandom_id": f2})
assert resp.status_code == 409
assert (await resp.get_json())["target"]["id"] == c2
# Confirm → merge c1 into c2, c2 survives.
resp = await client.patch(
f"/api/tags/{c1}", json={"fandom_id": f2, "merge": True}
)
assert resp.status_code == 200
assert (await resp.get_json())["id"] == c2
@pytest.mark.asyncio
async def test_rename_collision_returns_rich_409(client):
tgt = await _mk(client, "Canonical", "general")
+66 -1
View File
@@ -1,7 +1,11 @@
import pytest
from backend.app.models import TagKind
from backend.app.services.tag_service import TagService, TagValidationError
from backend.app.services.tag_service import (
TagMergeConflict,
TagService,
TagValidationError,
)
pytestmark = pytest.mark.integration
@@ -101,3 +105,64 @@ async def test_autocomplete_empty_query_returns_nothing(db):
svc = TagService(db)
await svc.find_or_create("Foo", TagKind.artist)
assert await svc.autocomplete("") == []
@pytest.mark.asyncio
async def test_set_fandom_assigns_changes_and_clears(db):
svc = TagService(db)
f1 = await svc.find_or_create("Bleach", TagKind.fandom)
f2 = await svc.find_or_create("Naruto", TagKind.fandom)
char = await svc.find_or_create("Ichigo", TagKind.character)
assert char.fandom_id is None
assert (await svc.set_fandom(char.id, f1.id)).fandom_id == f1.id
assert (await svc.set_fandom(char.id, f2.id)).fandom_id == f2.id
assert (await svc.set_fandom(char.id, None)).fandom_id is None
@pytest.mark.asyncio
async def test_set_fandom_rejects_non_character(db):
svc = TagService(db)
fandom = await svc.find_or_create("Naruto", TagKind.fandom)
series = await svc.find_or_create("Arc 1", TagKind.series)
with pytest.raises(TagValidationError, match="character"):
await svc.set_fandom(series.id, fandom.id)
@pytest.mark.asyncio
async def test_set_fandom_rejects_non_fandom_reference(db):
svc = TagService(db)
general = await svc.find_or_create("misc", TagKind.general)
char = await svc.find_or_create("Rukia", TagKind.character)
with pytest.raises(TagValidationError, match="fandom"):
await svc.set_fandom(char.id, general.id)
@pytest.mark.asyncio
async def test_set_fandom_collision_raises_merge_conflict(db):
svc = TagService(db)
f1 = await svc.find_or_create("Bleach", TagKind.fandom)
f2 = await svc.find_or_create("Naruto", TagKind.fandom)
c1 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f1.id)
c2 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f2.id)
with pytest.raises(TagMergeConflict) as ei:
await svc.set_fandom(c1.id, f2.id) # collides with c2 in f2
assert ei.value.target_id == c2.id
@pytest.mark.asyncio
async def test_set_fandom_merge_resolves_collision(db):
from sqlalchemy import select
from backend.app.models import Tag
svc = TagService(db)
f1 = await svc.find_or_create("Bleach", TagKind.fandom)
f2 = await svc.find_or_create("Naruto", TagKind.fandom)
c1 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f1.id)
c2 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f2.id)
survivor = await svc.set_fandom(c1.id, f2.id, merge=True)
assert survivor.id == c2.id # merged INTO the existing character
gone = (
await db.execute(select(Tag).where(Tag.id == c1.id))
).scalar_one_or_none()
assert gone is None