Senko-san
551afbab13
Thin adapters over the existing services/repositories (no business logic): - system: ping (auth check), getLicense - browsing: getArtists/getArtist/getAlbum, getAlbumList(2) (newest/alpha/random), getSong, getGenres, getMusicFolders/getIndexes/getMusicDirectory (one folder) - search: search3 (delegates to the library repos) - media: stream + download (reuse StreamingService, honor Range); getCoverArt returns a placeholder until the cover pipeline lands - playlists: get/create/update/delete over the playlist repo (owner-scoped) - annotation: star/unstar → append-only like log, scrobble → play history, setRating → clean no-op - all endpoints also accept the .view suffix and GET+POST for client compat Repo support: album list ordering (newest/random), track genre facets. README documents the mandatory-HTTPS requirement and app-password workflow. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
"""Album repository — adapter over ``AsyncSession``."""
|
|
|
|
import uuid
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.domain.entities.album import Album
|
|
from app.infrastructure.db.models.album import AlbumModel
|
|
from app.infrastructure.db.models.track import TrackModel
|
|
|
|
|
|
def _to_entity(row: AlbumModel) -> Album:
|
|
return Album(
|
|
id=row.id,
|
|
title=row.title,
|
|
artist_id=row.artist_id,
|
|
year=row.year,
|
|
cover_path=row.cover_path,
|
|
musicbrainz_id=row.musicbrainz_id,
|
|
created_at=row.created_at,
|
|
updated_at=row.updated_at,
|
|
)
|
|
|
|
|
|
class SqlAlchemyAlbumRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self._session = session
|
|
|
|
async def get_by_id(self, album_id: uuid.UUID) -> Album | None:
|
|
row = await self._session.get(AlbumModel, album_id)
|
|
return _to_entity(row) if row is not None else None
|
|
|
|
async def get_many(self, ids: list[uuid.UUID]) -> list[Album]:
|
|
if not ids:
|
|
return []
|
|
rows = (
|
|
(await self._session.execute(select(AlbumModel).where(AlbumModel.id.in_(ids))))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return [_to_entity(r) for r in rows]
|
|
|
|
async def count(self, *, artist_id: uuid.UUID | None, q: str | None) -> int:
|
|
stmt = select(func.count()).select_from(AlbumModel)
|
|
if artist_id is not None:
|
|
stmt = stmt.where(AlbumModel.artist_id == artist_id)
|
|
if q:
|
|
stmt = stmt.where(AlbumModel.title.ilike(f"%{q}%"))
|
|
return (await self._session.execute(stmt)).scalar_one()
|
|
|
|
async def track_count(self, album_id: uuid.UUID) -> int:
|
|
return (
|
|
await self._session.execute(
|
|
select(func.count()).select_from(TrackModel).where(TrackModel.album_id == album_id)
|
|
)
|
|
).scalar_one()
|
|
|
|
async def track_count_many(self, album_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]:
|
|
if not album_ids:
|
|
return {}
|
|
rows = (
|
|
await self._session.execute(
|
|
select(TrackModel.album_id, func.count(TrackModel.id).label("cnt"))
|
|
.where(TrackModel.album_id.in_(album_ids))
|
|
.group_by(TrackModel.album_id)
|
|
)
|
|
).all()
|
|
return {row.album_id: row.cnt for row in rows}
|
|
|
|
# list must come after methods using list[...] in signatures (builtin name shadowing)
|
|
async def list(
|
|
self,
|
|
*,
|
|
artist_id: uuid.UUID | None,
|
|
q: str | None,
|
|
limit: int,
|
|
offset: int,
|
|
sort_by: str = "title",
|
|
order: str = "asc",
|
|
) -> list[Album]:
|
|
stmt = select(AlbumModel)
|
|
if artist_id is not None:
|
|
stmt = stmt.where(AlbumModel.artist_id == artist_id)
|
|
if q:
|
|
stmt = stmt.where(AlbumModel.title.ilike(f"%{q}%"))
|
|
|
|
if order == "random":
|
|
stmt = stmt.order_by(func.random())
|
|
else:
|
|
col = AlbumModel.created_at if sort_by == "created" else AlbumModel.title
|
|
stmt = stmt.order_by(col.desc() if order == "desc" else col.asc())
|
|
stmt = stmt.limit(limit).offset(offset)
|
|
rows = (await self._session.execute(stmt)).scalars().all()
|
|
return [_to_entity(r) for r in rows]
|