Цвылев Александр Вадимович
a263272935
Implements previously-stubbed /api/v1 endpoints (hexagonal: ports -> repos
-> services -> routers wired in deps):
- playlists: GET /{id}/cover (serves stored cover, 404 when absent)
- settings: GET/PATCH /settings + GET/PUT /settings/scrobbling — lazy
per-user row, write-only Fernet-encrypted scrobble session key; adds
user_settings table + migration (chains off dc126696f5a6)
- storage: GET /duplicates, /broken, /missing-metadata + admin POST
/cleanup (arq cleanup_storage worker; reconciles local refs only,
guarded against a storage-outage mass delete)
- admin: GET /services, /sources, /settings + POST /reindex; PATCH
/settings and /sources/{source} return 501 (config is env-managed)
Adds NotSupportedError (-> HTTP 501). Integration tests for each surface.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
71 lines
2.9 KiB
Python
71 lines
2.9 KiB
Python
"""Source registry — selection + enumeration of configured backends.
|
|
|
|
Built from settings at the composition root. Only sources that are configured
|
|
are registered (e.g. ``local`` appears only when ``LOCAL_MEDIA_IMPORT_PATH`` is
|
|
set; ``youtube`` only when ``YOUTUBE_ENABLED``), so enumeration reflects what the
|
|
instance can actually use.
|
|
"""
|
|
|
|
from typing import cast
|
|
|
|
from app.core.config import Settings
|
|
from app.domain.errors import NotFoundError, ValidationError
|
|
from app.domain.ports import FetchableSource, IndexableSource, SearchableSource, SourceBackend
|
|
from app.domain.sources import SourceInfo
|
|
from app.infrastructure.sources.local_folder import LocalFolderSource
|
|
from app.infrastructure.sources.youtube import YouTubeMusicSource
|
|
|
|
|
|
class SourceRegistry:
|
|
def __init__(self, backends: list[SourceBackend]) -> None:
|
|
self._by_name = {backend.name: backend for backend in backends}
|
|
|
|
def get(self, name: str) -> SourceBackend:
|
|
backend = self._by_name.get(name)
|
|
if backend is None:
|
|
raise NotFoundError(f"Source {name!r} is not configured.")
|
|
return backend
|
|
|
|
def indexable(self, name: str) -> IndexableSource:
|
|
backend = self.get(name)
|
|
if not hasattr(backend, "scan"):
|
|
raise ValidationError(f"Source {name!r} cannot be indexed.")
|
|
return cast(IndexableSource, backend)
|
|
|
|
def searchable(self, name: str) -> SearchableSource:
|
|
backend = self.get(name)
|
|
if not hasattr(backend, "search"):
|
|
raise ValidationError(f"Source {name!r} cannot be searched.")
|
|
return cast(SearchableSource, backend)
|
|
|
|
def fetchable(self, name: str) -> FetchableSource:
|
|
backend = self.get(name)
|
|
if not hasattr(backend, "fetch"):
|
|
raise ValidationError(f"Source {name!r} cannot download.")
|
|
return cast(FetchableSource, backend)
|
|
|
|
def searchables(self) -> list[SearchableSource]:
|
|
"""Every registered source that supports search (for cross-source search)."""
|
|
return [cast(SearchableSource, b) for b in self._by_name.values() if hasattr(b, "search")]
|
|
|
|
def indexables(self) -> list[IndexableSource]:
|
|
"""Every registered source that can be indexed (for a full re-scan)."""
|
|
return [cast(IndexableSource, b) for b in self._by_name.values() if hasattr(b, "scan")]
|
|
|
|
def infos(self) -> list[SourceInfo]:
|
|
return [backend.info() for backend in self._by_name.values()]
|
|
|
|
|
|
def build_source_registry(settings: Settings) -> SourceRegistry:
|
|
backends: list[SourceBackend] = []
|
|
if settings.local_media_import_path is not None:
|
|
backends.append(LocalFolderSource(settings.local_media_import_path))
|
|
if settings.youtube_enabled:
|
|
backends.append(
|
|
YouTubeMusicSource(
|
|
cookies_path=settings.youtube_cookies_path,
|
|
tmp_dir=settings.upload_tmp_dir,
|
|
)
|
|
)
|
|
return SourceRegistry(backends)
|