refactor(dry-B2): ImportSettings.load()/load_sync() classmethods for the singleton row

The `select(ImportSettings).where(id == 1)).scalar_one()` singleton load was
repeated 15× across services, API, and 5 task modules. Added async load() +
sync load_sync() classmethods on the model and migrated all 15 full-row sites
(callers already imported ImportSettings, so no new imports; dropped download's
now-orphaned select import). Left maintenance.py's deliberate column-select
(import_scan_path only) as-is.

Rest of the service layer was already adequately DRY — the Record/to_dict
pattern is only 2 instances and the savepoint find-or-create recovery is
correctly per-entity, so neither was forced into a shared abstraction.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-28 14:03:26 -04:00
parent 21c1b0a81c
commit 171c486939
11 changed files with 26 additions and 47 deletions
+11 -1
View File
@@ -4,7 +4,7 @@ Enforced as a single row via a CHECK (id = 1) constraint. The application
always SELECTs id=1 and never inserts/deletes after the initial migration.
"""
from sqlalchemy import Boolean, CheckConstraint, Float, Integer, Text
from sqlalchemy import Boolean, CheckConstraint, Float, Integer, Text, select
from sqlalchemy.orm import Mapped, mapped_column
from .base import Base
@@ -63,3 +63,13 @@ class ImportSettings(Base):
backup_images_keep_last_n: Mapped[int] = mapped_column(
Integer, nullable=False, default=3,
)
@classmethod
async def load(cls, session) -> "ImportSettings":
"""The singleton settings row (id=1), via an async session."""
return (await session.execute(select(cls).where(cls.id == 1))).scalar_one()
@classmethod
def load_sync(cls, session) -> "ImportSettings":
"""The singleton settings row (id=1), via a sync session."""
return session.execute(select(cls).where(cls.id == 1)).scalar_one()