Merge pull request '#768 steps 1+2: normalized image_prediction table (read cutover)' (#92) from dev into main
CI / lint (push) Successful in 3s
Build images / sign-extension (push) Successful in 3s
CI / frontend-build (push) Successful in 22s
CI / backend-lint-and-test (push) Successful in 27s
Build images / build-web (push) Successful in 2m4s
Build images / build-ml (push) Successful in 2m42s
CI / integration (push) Successful in 3m8s

This commit was merged in pull request #92.
This commit is contained in:
2026-06-10 20:15:26 -04:00
12 changed files with 337 additions and 50 deletions
@@ -0,0 +1,82 @@
"""image_prediction table + backfill from image_record.tagger_predictions
Normalizes the per-image tagger predictions out of the JSON blob into a
queryable table (#768). Backfills from the existing JSON in one set-based
INSERT…SELECT over json_each — fast because the #764 prune already shrank
each row to its >=0.70 entries. The old image_record.tagger_predictions
column is left in place here (vestigial) and dropped in a follow-up once the
code cutover is verified — dropping it needs an ACCESS EXCLUSIVE lock on the
hot image_record table (the 0044 lock class), so it's deferred to a
quiesced-worker window.
Revision ID: 0045
Revises: 0044
Create Date: 2026-06-10
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "0045"
down_revision: Union[str, None] = "0044"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"image_prediction",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"image_record_id", sa.Integer(),
sa.ForeignKey("image_record.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("raw_name", sa.String(length=255), nullable=False),
sa.Column("category", sa.String(length=64), nullable=False),
sa.Column("score", sa.Float(), nullable=False),
sa.UniqueConstraint(
"image_record_id", "raw_name", name="image_raw_name",
),
)
op.create_index(
"ix_image_prediction_image", "image_prediction", ["image_record_id"],
)
op.create_index(
"ix_image_prediction_name_score", "image_prediction",
["raw_name", "score"],
)
# Backfill from the JSON blob. json_each expands {name: {category,
# confidence}} into one row per prediction. category defaults to 'general'
# to mirror the suggestion read path; rows with no confidence are skipped.
# Filter to >= the store floor (ml_settings.tagger_store_floor, default
# 0.70) right here so this is self-sufficient — it does NOT depend on the
# #764 prune having run, and extracting only the >=floor tail keeps
# image_prediction small (~tens of rows/image) even from the full JSON.
op.execute(
"""
INSERT INTO image_prediction (image_record_id, raw_name, category, score)
SELECT ir.id,
je.key,
COALESCE(je.value ->> 'category', 'general'),
(je.value ->> 'confidence')::double precision
FROM image_record ir,
json_each(ir.tagger_predictions) je
WHERE ir.tagger_predictions IS NOT NULL
AND je.value ->> 'confidence' IS NOT NULL
AND (je.value ->> 'confidence')::double precision
>= COALESCE(
(SELECT tagger_store_floor FROM ml_settings WHERE id = 1),
0.70
)
ON CONFLICT (image_record_id, raw_name) DO NOTHING
"""
)
def downgrade() -> None:
op.drop_index("ix_image_prediction_name_score", "image_prediction")
op.drop_index("ix_image_prediction_image", "image_prediction")
op.drop_table("image_prediction")
+2
View File
@@ -7,6 +7,7 @@ from .backup_run import BackupRun
from .base import Base
from .credential import Credential
from .download_event import DownloadEvent
from .image_prediction import ImagePrediction
from .image_provenance import ImageProvenance
from .image_record import ImageRecord
from .import_batch import ImportBatch
@@ -45,6 +46,7 @@ __all__ = [
"SeriesPage",
"SeriesSuggestion",
"ImageRecord",
"ImagePrediction",
"ImageProvenance",
"Tag",
"TagKind",
+37
View File
@@ -0,0 +1,37 @@
"""ImagePrediction — one row per (image, tagger vocab prediction).
Replaces the image_record.tagger_predictions JSON blob (#768). Storing the
raw Camie/booru vocab name (not a tag_id) preserves the suggestion read
path's semantics: raw_name → canonical Tag resolution happens at read time
via the alias map, and accepting a prediction can CREATE the Tag. The store
floor (ml_settings.tagger_store_floor) is applied at WRITE time, so only
predictions >= the floor land here.
"""
from sqlalchemy import Float, ForeignKey, Index, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from .base import Base
class ImagePrediction(Base):
__tablename__ = "image_prediction"
__table_args__ = (
UniqueConstraint(
"image_record_id", "raw_name", name="image_raw_name",
),
# Per-image read (suggestion build) and the "images with tag X above
# Y" query the JSON blob never allowed.
Index("ix_image_prediction_image", "image_record_id"),
Index("ix_image_prediction_name_score", "raw_name", "score"),
)
id: Mapped[int] = mapped_column(primary_key=True)
image_record_id: Mapped[int] = mapped_column(
ForeignKey("image_record.id", ondelete="CASCADE"), nullable=False,
)
# The raw tagger vocab key (booru form) — NOT a tag_id. Resolved to a
# canonical Tag at read time, exactly as the old JSON keys were.
raw_name: Mapped[str] = mapped_column(String(255), nullable=False)
category: Mapped[str] = mapped_column(String(64), nullable=False)
score: Mapped[float] = mapped_column(Float, nullable=False)
+10
View File
@@ -1090,6 +1090,16 @@ class Importer:
existing.siglip_embedding = None
existing.siglip_model_version = None
existing.centroid_scores = None
# #768: predictions also live in the normalized image_prediction table
# now — clear them so a re-imported file re-derives a fresh set.
from sqlalchemy import delete as _delete
from ..models import ImagePrediction as _ImagePrediction
self.session.execute(
_delete(_ImagePrediction).where(
_ImagePrediction.image_record_id == existing.id
)
)
# created_at intentionally preserved; updated_at auto-bumps.
self.session.flush()
self.session.commit()
+21 -1
View File
@@ -8,6 +8,7 @@ from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from ...models import (
ImagePrediction,
ImageRecord,
MLSettings,
Tag,
@@ -48,6 +49,25 @@ class SuggestionService:
await self.session.execute(select(MLSettings).where(MLSettings.id == 1))
).scalar_one()
async def _load_predictions(self, image_id: int) -> dict:
"""Predictions for one image from the normalized image_prediction
table (#768), in the {raw_name: {category, confidence}} shape the rest
of this service consumed from the old JSON column — so all downstream
threshold/alias/merge logic is unchanged."""
rows = (
await self.session.execute(
select(
ImagePrediction.raw_name,
ImagePrediction.category,
ImagePrediction.score,
).where(ImagePrediction.image_record_id == image_id)
)
).all()
return {
r.raw_name: {"category": r.category, "confidence": r.score}
for r in rows
}
def _threshold_for(
self, s: MLSettings, category: str, override: float | None = None,
) -> float:
@@ -80,7 +100,7 @@ class SuggestionService:
return SuggestionList()
settings = await self._settings()
predictions: dict = img.tagger_predictions or {}
predictions: dict = await self._load_predictions(image_id)
applied = set(
(
+46 -8
View File
@@ -10,11 +10,11 @@ import logging
from pathlib import Path
from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import select
from sqlalchemy import delete, select
from sqlalchemy.exc import DBAPIError, OperationalError
from ..celery_app import celery
from ..models import ImageRecord, MLSettings
from ..models import ImagePrediction, ImageRecord, MLSettings
from ._sync_engine import sync_session_factory as _sync_session_factory
log = logging.getLogger(__name__)
@@ -162,6 +162,23 @@ def tag_and_embed(self, image_id: int) -> dict:
record.siglip_embedding = embedding.tolist()
record.siglip_model_version = settings.embedder_model_version
session.add(record)
# Write the normalized image_prediction rows (#768). Delete-then-
# insert keeps a re-tag idempotent. tagger_store_floor was already
# applied in tagger.infer, so preds is the >=floor set. (Transitional
# dual-write alongside the JSON column until the read cutover lands.)
session.execute(
delete(ImagePrediction).where(
ImagePrediction.image_record_id == image_id
)
)
session.add_all([
ImagePrediction(
image_record_id=image_id, raw_name=name,
category=p.get("category", "general"),
score=float(p.get("confidence", 0.0)),
)
for name, p in preds.items()
])
session.commit()
except SoftTimeLimitExceeded:
log.error(
@@ -331,14 +348,16 @@ def apply_allowlist_tags(self, tag_id: int | None = None,
if not allow:
return 0
img_query = sa_select(
ImageRecord.id, ImageRecord.tagger_predictions
).where(ImageRecord.tagger_predictions.is_not(None))
# Images that have any predictions (#768: from image_prediction, not
# the old JSON column), optionally narrowed to one image.
img_ids_query = sa_select(ImagePrediction.image_record_id).distinct()
if image_id is not None:
img_query = img_query.where(ImageRecord.id == image_id)
img_ids_query = img_ids_query.where(
ImagePrediction.image_record_id == image_id
)
for img_id, preds in session.execute(img_query).all():
preds = preds or {}
for (img_id,) in session.execute(img_ids_query).all():
preds = _load_predictions_sync(session, img_id)
for a_tag_id, min_conf in allow.items():
exists = session.execute(
sa_select(image_tag.c.tag_id).where(
@@ -377,6 +396,25 @@ def apply_allowlist_tags(self, tag_id: int | None = None,
return applied
def _load_predictions_sync(session, image_id: int) -> dict:
"""Predictions for one image from image_prediction (#768), in the
{raw_name: {category, confidence}} shape _confidence_for_tag consumes —
keeps the allowlist resolution logic unchanged."""
from sqlalchemy import select as sa_select
rows = session.execute(
sa_select(
ImagePrediction.raw_name,
ImagePrediction.category,
ImagePrediction.score,
).where(ImagePrediction.image_record_id == image_id)
).all()
return {
r.raw_name: {"category": r.category, "confidence": r.score}
for r in rows
}
def _confidence_for_tag(session, tag, preds: dict) -> float | None:
"""Highest confidence among predictions that resolve to `tag` —
either the prediction name equals the tag name, or an alias maps
+21
View File
@@ -0,0 +1,21 @@
"""#768 test helper: seed image_prediction rows.
Read-path tests used to seed ImageRecord(tagger_predictions={...}); predictions
now live in the normalized image_prediction table, so seed there instead.
"""
from backend.app.models import ImagePrediction
async def seed_predictions(session, image_id: int, predictions: dict) -> None:
"""Insert image_prediction rows from a {raw_name: {category, confidence}}
dict (the old JSON shape). Caller commits/flushes as needed; this flushes."""
session.add_all([
ImagePrediction(
image_record_id=image_id,
raw_name=name,
category=p.get("category", "general"),
score=float(p.get("confidence", 0.0)),
)
for name, p in predictions.items()
])
await session.flush()
+4 -1
View File
@@ -15,14 +15,17 @@ def eager():
async def _img(db, preds):
from tests._prediction_helpers import seed_predictions
img = ImageRecord(
path="/images/s.jpg", sha256="s" * 64, size_bytes=1,
mime="image/jpeg", width=1, height=1,
origin="imported_filesystem", integrity_status="unknown",
tagger_predictions=preds,
)
db.add(img)
await db.commit()
await seed_predictions(db, img.id, preds)
await db.commit()
return img
+57
View File
@@ -0,0 +1,57 @@
"""#768: image_prediction table — model + constraints round-trip."""
import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from backend.app.models import ImagePrediction, ImageRecord
pytestmark = pytest.mark.integration
async def _make_image(db, path="/img/p0.jpg", sha="0"):
rec = ImageRecord(
path=path, sha256=sha.ljust(64, "0")[:64], size_bytes=10,
mime="image/jpeg", origin="imported_filesystem",
)
db.add(rec)
await db.flush()
return rec
@pytest.mark.asyncio
async def test_image_prediction_round_trip(db):
rec = await _make_image(db)
db.add_all([
ImagePrediction(
image_record_id=rec.id, raw_name="blue_eyes",
category="general", score=0.92,
),
ImagePrediction(
image_record_id=rec.id, raw_name="hatsune_miku",
category="character", score=0.88,
),
])
await db.commit()
rows = (await db.execute(
select(ImagePrediction.raw_name, ImagePrediction.score)
.where(ImagePrediction.image_record_id == rec.id)
.order_by(ImagePrediction.score.desc())
)).all()
assert [r.raw_name for r in rows] == ["blue_eyes", "hatsune_miku"]
@pytest.mark.asyncio
async def test_image_prediction_unique_per_image_name(db):
rec = await _make_image(db, path="/img/p1.jpg", sha="1")
db.add(ImagePrediction(
image_record_id=rec.id, raw_name="dup",
category="general", score=0.9,
))
await db.commit()
db.add(ImagePrediction(
image_record_id=rec.id, raw_name="dup",
category="general", score=0.7,
))
with pytest.raises(IntegrityError):
await db.commit()
+28 -23
View File
@@ -9,7 +9,7 @@ from backend.app.services.tag_service import TagService
pytestmark = pytest.mark.integration
def _img(sha: str, predictions: dict) -> ImageRecord:
def _img(sha: str) -> ImageRecord:
return ImageRecord(
path=f"/images/{sha}.jpg",
sha256=sha,
@@ -19,24 +19,34 @@ def _img(sha: str, predictions: dict) -> ImageRecord:
height=1,
origin="imported_filesystem",
integrity_status="unknown",
tagger_predictions=predictions,
)
async def _seed_img(db, sha: str, predictions: dict) -> ImageRecord:
"""#768: create an image + seed its predictions into image_prediction
(the read path's source), returning the flushed record."""
from tests._prediction_helpers import seed_predictions
img = _img(sha)
db.add(img)
await db.flush()
await seed_predictions(db, img.id, predictions)
return img
@pytest.mark.asyncio
async def test_threshold_filters_low_confidence_general(db):
# Default general threshold is 0.50 (alembic 0029 lowered it from
# 0.95). Use 0.30/0.60 to keep the test asserting threshold behavior
# rather than the exact cutoff number.
img = _img(
img = await _seed_img(
db,
"a" * 64,
{
"lowconf": {"category": "general", "confidence": 0.30},
"sword": {"category": "general", "confidence": 0.97},
},
)
db.add(img)
await db.flush()
sl = await SuggestionService(db).for_image(img.id)
names = [s.display_name for s in sl.by_category.get("general", [])]
# display_name is normalized (tag_name.normalize) before surfacing.
@@ -49,36 +59,34 @@ async def test_threshold_override_surfaces_low_confidence(db):
# The typed-dropdown "show everything the model saw" mode: threshold_override
# surfaces stored predictions below the configured threshold (in canonical
# formatting) so they can be picked instead of hand-typed (2026-06-09).
img = _img(
img = await _seed_img(
db,
"d" * 64,
{
"lowconf": {"category": "general", "confidence": 0.30},
"sword": {"category": "general", "confidence": 0.97},
},
)
db.add(img)
await db.flush()
sl = await SuggestionService(db).for_image(img.id, threshold_override=0.0)
names = [s.display_name for s in sl.by_category.get("general", [])]
assert "Sword" in names
assert "Lowconf" in names # below the configured threshold, surfaced anyway
# Unsurfaced categories are still excluded even with the override.
img2 = _img("e" * 64, {"safe": {"category": "rating", "confidence": 0.99}})
db.add(img2)
await db.flush()
img2 = await _seed_img(
db, "e" * 64, {"safe": {"category": "rating", "confidence": 0.99}}
)
sl2 = await SuggestionService(db).for_image(img2.id, threshold_override=0.0)
assert "rating" not in sl2.by_category
@pytest.mark.asyncio
async def test_unsurfaced_category_dropped(db):
img = _img(
img = await _seed_img(
db,
"b" * 64,
{"safe": {"category": "rating", "confidence": 0.99}},
)
db.add(img)
await db.flush()
sl = await SuggestionService(db).for_image(img.id)
assert "rating" not in sl.by_category
@@ -88,12 +96,11 @@ async def test_alias_resolution(db):
tags = TagService(db)
canonical = await tags.find_or_create("Sasuke Uchiha", TagKind.character)
await AliasService(db).create("uchiha_sasuke", "character", canonical.id)
img = _img(
img = await _seed_img(
db,
"c" * 64,
{"uchiha_sasuke": {"category": "character", "confidence": 0.96}},
)
db.add(img)
await db.flush()
sl = await SuggestionService(db).for_image(img.id)
chars = sl.by_category["character"]
assert len(chars) == 1
@@ -104,12 +111,11 @@ async def test_alias_resolution(db):
@pytest.mark.asyncio
async def test_raw_tag_creates_new(db):
img = _img(
img = await _seed_img(
db,
"d" * 64,
{"brand_new_tag": {"category": "character", "confidence": 0.96}},
)
db.add(img)
await db.flush()
sl = await SuggestionService(db).for_image(img.id)
chars = sl.by_category["character"]
# display_name is the normalized Camie name (underscores -> spaces,
@@ -123,12 +129,11 @@ async def test_raw_tag_creates_new(db):
async def test_applied_tag_not_suggested(db):
tags = TagService(db)
tag = await tags.find_or_create("alreadyhere", TagKind.character)
img = _img(
img = await _seed_img(
db,
"e" * 64,
{"alreadyhere": {"category": "character", "confidence": 0.96}},
)
db.add(img)
await db.flush()
await db.execute(
image_tag.insert().values(
image_record_id=img.id, tag_id=tag.id, source="manual"
+19 -11
View File
@@ -5,16 +5,16 @@ from backend.app.models import ImageRecord, TagKind
from backend.app.models.tag import image_tag
from backend.app.services.ml.suggestions import SuggestionService
from backend.app.services.tag_service import TagService
from tests._prediction_helpers import seed_predictions
pytestmark = pytest.mark.integration
def _img(sha: str, predictions: dict) -> ImageRecord:
def _img(sha: str) -> ImageRecord:
return ImageRecord(
path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1,
mime="image/jpeg", width=1, height=1,
origin="imported_filesystem", integrity_status="unknown",
tagger_predictions=predictions,
)
@@ -22,10 +22,12 @@ def _img(sha: str, predictions: dict) -> ImageRecord:
async def test_consensus_includes_tag_over_threshold(db):
tags = TagService(db)
t = await tags.find_or_create("sword", TagKind.general)
a = _img("a" * 64, {"sword": {"category": "general", "confidence": 0.97}})
b = _img("b" * 64, {"sword": {"category": "general", "confidence": 0.95}})
a = _img("a" * 64)
b = _img("b" * 64)
db.add_all([a, b])
await db.flush()
await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}})
await seed_predictions(db, b.id, {"sword": {"category": "general", "confidence": 0.95}})
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
gen = res["general"]
assert any(s["canonical_tag_id"] == t.id for s in gen)
@@ -38,10 +40,11 @@ async def test_consensus_includes_tag_over_threshold(db):
async def test_consensus_counts_already_applied_for_coverage(db):
tags = TagService(db)
t = await tags.find_or_create("sky", TagKind.general)
a = _img("c" * 64, {"sky": {"category": "general", "confidence": 0.96}})
b = _img("d" * 64, {}) # no prediction
a = _img("c" * 64)
b = _img("d" * 64) # no prediction
db.add_all([a, b])
await db.flush()
await seed_predictions(db, a.id, {"sky": {"category": "general", "confidence": 0.96}})
# b already has the tag applied -> counts toward coverage, not confidence
await db.execute(
image_tag.insert().values(
@@ -58,10 +61,11 @@ async def test_consensus_counts_already_applied_for_coverage(db):
async def test_consensus_excludes_below_threshold(db):
tags = TagService(db)
await tags.find_or_create("rare", TagKind.general)
a = _img("e" * 64, {"rare": {"category": "general", "confidence": 0.96}})
b = _img("f" * 64, {})
a = _img("e" * 64)
b = _img("f" * 64)
db.add_all([a, b])
await db.flush()
await seed_predictions(db, a.id, {"rare": {"category": "general", "confidence": 0.96}})
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
assert all(
s["name"] != "rare" for s in res.get("general", [])
@@ -70,10 +74,12 @@ async def test_consensus_excludes_below_threshold(db):
@pytest.mark.asyncio
async def test_consensus_skips_creates_new_tag(db):
a = _img("g" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
b = _img("h" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
a = _img("g" * 64)
b = _img("h" * 64)
db.add_all([a, b])
await db.flush()
await seed_predictions(db, a.id, {"neverseen": {"category": "general", "confidence": 0.99}})
await seed_predictions(db, b.id, {"neverseen": {"category": "general", "confidence": 0.99}})
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
# 'neverseen' has no Tag row -> creates_new_tag -> excluded from consensus
assert all(s["name"] != "neverseen" for s in res.get("general", []))
@@ -90,9 +96,11 @@ async def test_bulk_suggestions_route(db):
tags = TagService(db)
await tags.find_or_create("sword", TagKind.general)
a = _img("i" * 64, {"sword": {"category": "general", "confidence": 0.97}})
a = _img("i" * 64)
db.add(a)
await db.commit()
await seed_predictions(db, a.id, {"sword": {"category": "general", "confidence": 0.97}})
await db.commit()
app = create_app()
async with app.test_client() as c:
resp = await c.post(
+10 -6
View File
@@ -63,6 +63,7 @@ async def test_apply_allowlist_applies_above_threshold(db):
from backend.app.models.tag import image_tag
from backend.app.services.tag_service import TagService
from backend.app.tasks import ml as ml_tasks
from tests._prediction_helpers import seed_predictions
tag = await TagService(db).find_or_create("autohero", TagKind.character)
db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
@@ -70,12 +71,13 @@ async def test_apply_allowlist_applies_above_threshold(db):
path="/images/al.jpg", sha256="al" + "0" * 62, size_bytes=1,
mime="image/jpeg", width=1, height=1,
origin="imported_filesystem", integrity_status="unknown",
tagger_predictions={
"autohero": {"category": "character", "confidence": 0.97}
},
)
db.add(img)
await db.commit()
await seed_predictions(
db, img.id, {"autohero": {"category": "character", "confidence": 0.97}}
)
await db.commit()
n = ml_tasks.apply_allowlist_tags(tag_id=tag.id)
assert n >= 1
@@ -98,6 +100,7 @@ async def test_apply_allowlist_skips_below_threshold(db):
from backend.app.models.tag import image_tag
from backend.app.services.tag_service import TagService
from backend.app.tasks import ml as ml_tasks
from tests._prediction_helpers import seed_predictions
tag = await TagService(db).find_or_create("lowconf", TagKind.character)
db.add(TagAllowlist(tag_id=tag.id, min_confidence=0.95))
@@ -105,12 +108,13 @@ async def test_apply_allowlist_skips_below_threshold(db):
path="/images/lc.jpg", sha256="lc" + "0" * 62, size_bytes=1,
mime="image/jpeg", width=1, height=1,
origin="imported_filesystem", integrity_status="unknown",
tagger_predictions={
"lowconf": {"category": "character", "confidence": 0.40}
},
)
db.add(img)
await db.commit()
await seed_predictions(
db, img.id, {"lowconf": {"category": "character", "confidence": 0.40}}
)
await db.commit()
ml_tasks.apply_allowlist_tags(tag_id=tag.id)
applied = (
await db.execute(