"""FC-3a: CRUD over Source rows, with platform/url/config validation and is_subscription auto-flip on first add / last delete. """ from dataclasses import dataclass from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..models import Artist, ImportSettings, Source from .platforms import known_platform_keys from .scheduler_service import compute_next_check_at # Re-exported for back-compat with FC-3a; the registry in # services/platforms.py is the single source of truth. KNOWN_PLATFORMS = known_platform_keys() # --- Errors ----------------------------------------------------------------- class SourceServiceError(Exception): """Base.""" class ArtistNotFoundError(SourceServiceError): pass class UnknownPlatformError(SourceServiceError): def __init__(self, platform: str): super().__init__(f"unknown platform: {platform!r}") self.platform = platform class InvalidConfigError(SourceServiceError): pass class EmptyUrlError(SourceServiceError): pass class DuplicateSourceError(SourceServiceError): def __init__(self, existing_id: int): super().__init__(f"duplicate source (existing id={existing_id})") self.existing_id = existing_id @dataclass(frozen=True) class SourceRecord: id: int artist_id: int artist_name: str artist_slug: str platform: str url: str enabled: bool config_overrides: dict | None last_checked_at: str | None last_error: str | None check_interval_override: int | None consecutive_failures: int next_check_at: str | None def to_dict(self) -> dict: return { "id": self.id, "artist_id": self.artist_id, "artist_name": self.artist_name, "artist_slug": self.artist_slug, "platform": self.platform, "url": self.url, "enabled": self.enabled, "config_overrides": self.config_overrides, "last_checked_at": self.last_checked_at, "last_error": self.last_error, "check_interval_override": self.check_interval_override, "consecutive_failures": self.consecutive_failures, "next_check_at": self.next_check_at, } # --- Service ---------------------------------------------------------------- _EDITABLE = {"enabled", "url", "config_overrides", "check_interval_override", "platform"} class SourceService: def __init__(self, session: AsyncSession): self.session = session async def _artist_or_raise(self, artist_id: int) -> Artist: row = (await self.session.execute( select(Artist).where(Artist.id == artist_id) )).scalar_one_or_none() if row is None: raise ArtistNotFoundError(f"artist id={artist_id} not found") return row @staticmethod def _validate_platform(platform: str) -> str: if platform not in KNOWN_PLATFORMS: raise UnknownPlatformError(platform) return platform @staticmethod def _validate_url(url: str) -> str: cleaned = (url or "").strip() if not cleaned: raise EmptyUrlError("url must not be empty") return cleaned @staticmethod def _validate_config(config: object) -> dict | None: if config is None: return None if not isinstance(config, dict): raise InvalidConfigError("config_overrides must be a JSON object") return config async def _load_settings(self) -> ImportSettings: return await ImportSettings.load(self.session) def _build_record( self, source: Source, artist: Artist, settings: ImportSettings, ) -> SourceRecord: nxt = compute_next_check_at(source, artist, settings) return SourceRecord( id=source.id, artist_id=source.artist_id, artist_name=artist.name, artist_slug=artist.slug, platform=source.platform, url=source.url, enabled=source.enabled, config_overrides=source.config_overrides, last_checked_at=source.last_checked_at.isoformat() if source.last_checked_at else None, last_error=source.last_error, check_interval_override=source.check_interval_override, consecutive_failures=source.consecutive_failures or 0, next_check_at=nxt.isoformat() if nxt else None, ) async def _row_to_record(self, source: Source) -> SourceRecord: artist = (await self.session.execute( select(Artist).where(Artist.id == source.artist_id) )).scalar_one() settings = await self._load_settings() return self._build_record(source, artist, settings) async def list( self, artist_id: int | None = None, failing: bool = False, include_synthetic: bool = False, ) -> list[SourceRecord]: stmt = select(Source, Artist).join(Artist, Artist.id == Source.artist_id) if artist_id is not None: stmt = stmt.where(Source.artist_id == artist_id) if not include_synthetic: # Pre-alembic-0030 sidecar synthetic anchors # have url='sidecar::' and exist only to give # imported Posts a NOT-NULL Source FK. They aren't pollable # feeds; the Subscriptions UI used to render them as phantom # subscriptions. Hide by default. stmt = stmt.where(~Source.url.like("sidecar:%")) if failing: # Worst-first so the rollup card surfaces the loudest failures. stmt = stmt.where(Source.consecutive_failures > 0).order_by( Source.consecutive_failures.desc(), Artist.name.asc(), ) else: stmt = stmt.order_by(Artist.name.asc(), Source.id.asc()) rows = (await self.session.execute(stmt)).all() settings = await self._load_settings() return [self._build_record(s, a, settings) for s, a in rows] async def get(self, source_id: int) -> SourceRecord | None: source = (await self.session.execute( select(Source).where(Source.id == source_id) )).scalar_one_or_none() if source is None: return None return await self._row_to_record(source) async def create( self, *, artist_id: int, platform: str, url: str, enabled: bool = True, config_overrides: dict | None = None, check_interval_override: int | None = None, ) -> SourceRecord: await self._artist_or_raise(artist_id) platform = self._validate_platform(platform) url = self._validate_url(url) config_overrides = self._validate_config(config_overrides) prior_count = (await self.session.execute( select(func.count(Source.id)).where(Source.artist_id == artist_id) )).scalar_one() source = Source( artist_id=artist_id, platform=platform, url=url, enabled=enabled, config_overrides=config_overrides, check_interval_override=check_interval_override, ) self.session.add(source) try: await self.session.flush() except IntegrityError as exc: await self.session.rollback() existing = (await self.session.execute( select(Source.id).where( Source.artist_id == artist_id, Source.platform == platform, Source.url == url, ) )).scalar_one() raise DuplicateSourceError(existing_id=existing) from exc if prior_count == 0: await self.session.execute( Artist.__table__.update() .where(Artist.id == artist_id) .values(is_subscription=True) ) await self.session.commit() return await self._row_to_record(source) async def update(self, source_id: int, **fields) -> SourceRecord: source = (await self.session.execute( select(Source).where(Source.id == source_id) )).scalar_one_or_none() if source is None: raise LookupError(f"source id={source_id} not found") for key in list(fields.keys()): if key not in _EDITABLE: fields.pop(key) if "platform" in fields: fields["platform"] = self._validate_platform(fields["platform"]) if "url" in fields: fields["url"] = self._validate_url(fields["url"]) if "config_overrides" in fields: fields["config_overrides"] = self._validate_config(fields["config_overrides"]) for key, value in fields.items(): setattr(source, key, value) try: await self.session.flush() except IntegrityError as exc: await self.session.rollback() existing = (await self.session.execute( select(Source.id).where( Source.artist_id == source.artist_id, Source.platform == source.platform, Source.url == source.url, Source.id != source.id, ) )).scalar_one() raise DuplicateSourceError(existing_id=existing) from exc await self.session.commit() return await self._row_to_record(source) async def delete(self, source_id: int) -> None: source = (await self.session.execute( select(Source).where(Source.id == source_id) )).scalar_one_or_none() if source is None: raise LookupError(f"source id={source_id} not found") artist_id = source.artist_id await self.session.delete(source) await self.session.flush() remaining = (await self.session.execute( select(func.count(Source.id)).where(Source.artist_id == artist_id) )).scalar_one() if remaining == 0: await self.session.execute( Artist.__table__.update() .where(Artist.id == artist_id) .values(is_subscription=False) ) await self.session.commit()