fix(database): preserve pair tags during global maintenance (#193)
This commit is contained in:
Vito Sansevero
+26
-15
@@ -924,6 +924,8 @@ class PromptDatabase(PairOperations):
|
||||
|
||||
old_name = old_name.strip()
|
||||
new_name = new_name.strip()
|
||||
if old_name == new_name:
|
||||
return {"success": True, "affected_count": 0, "skipped_count": 0}
|
||||
|
||||
with self.model.get_connection() as conn:
|
||||
cursor = conn.execute("SELECT id FROM tags WHERE name = ?", (old_name,))
|
||||
@@ -945,12 +947,14 @@ class PromptDatabase(PairOperations):
|
||||
if existing_new:
|
||||
# Target tag exists — merge: move links, handle conflicts, delete old
|
||||
new_tag_id = existing_new["id"]
|
||||
conn.execute(
|
||||
"UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?",
|
||||
(new_tag_id, old_tag_id),
|
||||
)
|
||||
# Delete remaining links (conflicts = prompt already had the target tag)
|
||||
conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (old_tag_id,))
|
||||
for junction in ("prompt_tags", "prompt_pair_tags"):
|
||||
conn.execute(
|
||||
f"UPDATE OR IGNORE {junction} SET tag_id = ? WHERE tag_id = ?",
|
||||
(new_tag_id, old_tag_id),
|
||||
)
|
||||
conn.execute(
|
||||
f"DELETE FROM {junction} WHERE tag_id = ?", (old_tag_id,)
|
||||
)
|
||||
conn.execute("DELETE FROM tags WHERE id = ?", (old_tag_id,))
|
||||
else:
|
||||
# Simple rename
|
||||
@@ -969,7 +973,7 @@ class PromptDatabase(PairOperations):
|
||||
"""
|
||||
Remove a tag from all prompts that use it.
|
||||
|
||||
With junction tables, deleting from the tags table cascades to prompt_tags.
|
||||
Deleting from tags cascades to standalone and pair tag relationships.
|
||||
|
||||
Args:
|
||||
tag_name: Tag name to remove
|
||||
@@ -1036,20 +1040,27 @@ class PromptDatabase(PairOperations):
|
||||
if not src_row:
|
||||
continue
|
||||
src_id = src_row["id"]
|
||||
if src_id == target_id:
|
||||
continue
|
||||
|
||||
cursor = conn.execute(
|
||||
"SELECT COUNT(*) as c FROM prompt_tags WHERE tag_id = ?", (src_id,)
|
||||
)
|
||||
src_count = cursor.fetchone()["c"]
|
||||
|
||||
if src_count > 0:
|
||||
# Move links from source to target (ignore conflicts)
|
||||
conn.execute(
|
||||
"UPDATE OR IGNORE prompt_tags SET tag_id = ? WHERE tag_id = ?",
|
||||
(target_id, src_id),
|
||||
)
|
||||
# Delete remaining (conflicts = prompt already had target tag)
|
||||
conn.execute("DELETE FROM prompt_tags WHERE tag_id = ?", (src_id,))
|
||||
pair_count = conn.execute(
|
||||
"SELECT COUNT(*) FROM prompt_pair_tags WHERE tag_id = ?", (src_id,)
|
||||
).fetchone()[0]
|
||||
if src_count or pair_count:
|
||||
for junction in ("prompt_tags", "prompt_pair_tags"):
|
||||
conn.execute(
|
||||
f"UPDATE OR IGNORE {junction} "
|
||||
"SET tag_id = ? WHERE tag_id = ?",
|
||||
(target_id, src_id),
|
||||
)
|
||||
conn.execute(
|
||||
f"DELETE FROM {junction} WHERE tag_id = ?", (src_id,)
|
||||
)
|
||||
tags_merged += 1
|
||||
affected += src_count
|
||||
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Dependency checks exclude competing pair writers until mutation finishes."""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest.mock
|
||||
|
||||
from database.operations import PromptDatabase
|
||||
from utils.hashing import generate_prompt_hash
|
||||
|
||||
|
||||
class TestPairDependencyConcurrency(unittest.TestCase):
|
||||
def test_legacy_mutators_lock_out_pair_creation_before_dependency_checks(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = os.path.join(directory, "concurrent.db")
|
||||
db = PromptDatabase(path)
|
||||
try:
|
||||
original = db._ensure_prompts_unpaired
|
||||
checks = []
|
||||
|
||||
def check_with_competing_writer(conn, ids):
|
||||
original(conn, ids)
|
||||
competitor = sqlite3.connect(path, timeout=0)
|
||||
try:
|
||||
with self.assertRaisesRegex(sqlite3.OperationalError, "locked"):
|
||||
competitor.execute("BEGIN IMMEDIATE")
|
||||
finally:
|
||||
competitor.close()
|
||||
checks.append(list(ids))
|
||||
|
||||
actions = (
|
||||
lambda pid: db.update_prompt_text(pid, "changed"),
|
||||
db.delete_prompt,
|
||||
lambda pid: db.bulk_delete_prompts([pid]),
|
||||
lambda _pid: db.delete_prompts_by_category("temporary"),
|
||||
)
|
||||
with unittest.mock.patch.object(
|
||||
db, "_ensure_prompts_unpaired", check_with_competing_writer
|
||||
):
|
||||
for index, action in enumerate(actions):
|
||||
with self.subTest(action=index):
|
||||
text = f"unpaired {index}"
|
||||
pid = db.save_prompt(
|
||||
text,
|
||||
category="temporary",
|
||||
prompt_hash=generate_prompt_hash(text),
|
||||
)
|
||||
self.assertTrue(action(pid))
|
||||
self.assertEqual(len(checks), len(actions))
|
||||
finally:
|
||||
db.close_all()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Global tag maintenance preserves pair and standalone tag relationships."""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from database.operations import PromptDatabase
|
||||
|
||||
|
||||
class TestPairTagMaintenance(unittest.TestCase):
|
||||
def setUp(self):
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
self.db = PromptDatabase(os.path.join(directory.name, "pair-tags.db"))
|
||||
self.addCleanup(self.db.close_all)
|
||||
|
||||
def pair(self, positive: str, tags: list) -> int:
|
||||
return self.db.save_prompt_pair(positive, "blurry", tags=tags)
|
||||
|
||||
def tags(self, pair_id: int) -> list:
|
||||
return self.db.get_prompt_pair(pair_id)["tags"]
|
||||
|
||||
def test_rename_to_existing_tag_preserves_pair_links_and_deduplicates(self):
|
||||
first = self.pair("portrait", ["old", "new", "keep"])
|
||||
second = self.pair("landscape", ["old"])
|
||||
third = self.pair("studio", ["unrelated"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["old", "new"])
|
||||
result = self.db.rename_tag_all_prompts("old", "new")
|
||||
self.assertEqual(result["affected_count"], 1)
|
||||
self.assertEqual(self.tags(first), ["keep", "new"])
|
||||
self.assertEqual(self.tags(second), ["new"])
|
||||
self.assertEqual(self.tags(third), ["unrelated"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["new"])
|
||||
self.assertEqual(self.db.list_prompt_pairs(tags=["new"])["total"], 2)
|
||||
self.assertNotIn("old", self.db.get_all_tags())
|
||||
|
||||
def test_rename_pair_only_tag_to_new_name_preserves_links(self):
|
||||
pair = self.pair("portrait", ["old"])
|
||||
self.db.rename_tag_all_prompts("old", "new")
|
||||
self.assertEqual(self.tags(pair), ["new"])
|
||||
|
||||
def test_merge_pair_only_source_preserves_its_links(self):
|
||||
pair = self.pair("portrait", ["source"])
|
||||
other = self.pair("landscape", ["target"])
|
||||
self.db.merge_tags(["source"], "target")
|
||||
self.assertEqual(self.tags(pair), ["target"])
|
||||
self.assertEqual(self.tags(other), ["target"])
|
||||
self.assertEqual(self.db.list_prompt_pairs(tags=["target"])["total"], 2)
|
||||
|
||||
def test_merge_multiple_sources_preserves_shared_and_unrelated_tags(self):
|
||||
pair = self.pair("portrait", ["first", "second", "target", "keep"])
|
||||
other = self.pair("landscape", ["second"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["first", "second"])
|
||||
self.db.merge_tags(["first", "second"], "target")
|
||||
self.assertEqual(self.tags(pair), ["keep", "target"])
|
||||
self.assertEqual(self.tags(other), ["target"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["target"])
|
||||
self.assertEqual(self.db.list_prompt_pairs(tags=["target"])["total"], 2)
|
||||
|
||||
def test_self_rename_does_not_remove_tag_or_relationships(self):
|
||||
pair = self.pair("portrait", ["same"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["same"])
|
||||
self.db.rename_tag_all_prompts("same", "same")
|
||||
self.assertEqual(self.tags(pair), ["same"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["same"])
|
||||
|
||||
def test_merge_skips_target_even_when_it_is_in_source_list(self):
|
||||
pair = self.pair("portrait", ["target", "source"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["target", "source"])
|
||||
self.db.merge_tags(["target", "source", "target"], "target")
|
||||
self.assertEqual(self.tags(pair), ["target"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["target"])
|
||||
self.assertIn("target", self.db.get_all_tags())
|
||||
|
||||
def test_explicit_global_delete_removes_pair_tags_and_keeps_other_tags(self):
|
||||
pair = self.pair("portrait", ["remove", "keep"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["remove", "keep"])
|
||||
self.db.delete_tag_all_prompts("remove")
|
||||
self.assertEqual(self.tags(pair), ["keep"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["keep"])
|
||||
self.assertIsNotNone(self.db.get_prompt_pair(pair))
|
||||
|
||||
def test_pair_link_failure_rolls_back_standalone_merge_and_source_delete(self):
|
||||
pair = self.pair("portrait", ["source"])
|
||||
prompt = self.db.save_prompt("standalone", tags=["source"])
|
||||
conn = self.db.model.get_connection()
|
||||
conn.execute(
|
||||
"CREATE TRIGGER reject_pair_tag_change BEFORE UPDATE "
|
||||
"ON prompt_pair_tags BEGIN SELECT RAISE(ABORT,'pair tag failure'); END"
|
||||
)
|
||||
conn.commit()
|
||||
with self.assertRaises(sqlite3.IntegrityError):
|
||||
self.db.merge_tags(["source"], "target")
|
||||
self.assertEqual(self.tags(pair), ["source"])
|
||||
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["source"])
|
||||
self.assertNotIn("target", self.db.get_all_tags())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user