feat: implement 1B domain entities/repos + 1H library API routes

1B — domain layer:
- New entities: Album, Playlist, Like, PlayHistoryEntry
- Track entity extended with album_id, genre, year fields
- New protocols: AlbumRepository, PlaylistRepository, LikeRepository, HistoryRepository
- ArtistRepository / TrackRepository protocols extended (list, count, update, get_many, etc.)
- New repos: SqlAlchemyAlbum/Playlist/Like/HistoryRepository
- Artist and track repos updated to match extended protocols

1H — library API:
- Pagination: PagedResponse[T] generic, offset-based, limit default 50 max 200
- Schemas: TrackOut, AlbumOut, ArtistOut, PlaylistOut/Create/Update,
  LikeEvent/State, HistoryIn/Out, LibrarySearchResponse
- GET/PATCH/DELETE /tracks with filters, sort, pagination
- GET /albums, /albums/{id}, /albums/{id}/tracks
- GET /artists, /artists/{id}, /artists/{id}/albums, /artists/{id}/tracks
- GET /search/library (ILIKE across tracks/albums/artists)
- Full /playlists CRUD + track add/remove (append-only version bump)
- POST /likes (append-only event log), GET /likes, GET /likes/state
- POST /history (scrobble), GET /history
- deps.py: TrackRepoDep, ArtistRepoDep, AlbumRepoDep, PlaylistRepoDep,
  LikeRepoDep, HistoryRepoDep

ruff   mypy   pytest 45/45 

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Senko-san
2026-06-07 16:43:51 +03:00
parent 81ea93c371
commit 7c920f38f6
30 changed files with 1641 additions and 59 deletions
+93 -5
View File
@@ -3,22 +3,110 @@
import uuid
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Query
from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, TrackRepoDep
from app.api.schemas.album import AlbumOut
from app.api.schemas.pagination import PagedResponse
from app.api.schemas.track import TrackOut
from app.api.v1.tracks import _build_track_out
from app.domain.entities.album import Album
from app.domain.entities.track import Artist
from app.domain.errors import NotFoundError
router = APIRouter(prefix="/albums", tags=["albums"])
async def _build_album_out(
albums: list[Album],
artists: dict[uuid.UUID, Artist],
track_counts: dict[uuid.UUID, int],
) -> list[AlbumOut]:
return [
AlbumOut(
id=a.id,
title=a.title,
artist_id=a.artist_id,
artist_name=artists[a.artist_id].name if a.artist_id in artists else "Unknown Artist",
year=a.year,
track_count=track_counts.get(a.id, 0),
created_at=a.created_at,
)
for a in albums
]
@router.get("")
async def list_albums() -> Any: ...
async def list_albums(
album_repo: AlbumRepoDep,
artist_repo: ArtistRepoDep,
_: CurrentUser,
artist_id: uuid.UUID | None = None,
q: str | None = None,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[AlbumOut]:
albums = await album_repo.list(artist_id=artist_id, q=q, limit=limit, offset=offset)
total = await album_repo.count(artist_id=artist_id, q=q)
artist_ids = list({a.artist_id for a in albums})
artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
track_counts = await album_repo.track_count_many([a.id for a in albums])
items = await _build_album_out(albums, artists, track_counts)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{album_id}")
async def get_album(album_id: uuid.UUID) -> Any: ...
async def get_album(
album_id: uuid.UUID,
album_repo: AlbumRepoDep,
artist_repo: ArtistRepoDep,
_: CurrentUser,
) -> AlbumOut:
album = await album_repo.get_by_id(album_id)
if album is None:
raise NotFoundError(f"Album {album_id} not found.")
artists = {a.id: a for a in await artist_repo.get_many([album.artist_id])}
track_counts = await album_repo.track_count_many([album.id])
items = await _build_album_out([album], artists, track_counts)
return items[0]
@router.get("/{album_id}/tracks")
async def get_album_tracks(album_id: uuid.UUID) -> Any: ...
async def get_album_tracks(
album_id: uuid.UUID,
track_repo: TrackRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[TrackOut]:
album = await album_repo.get_by_id(album_id)
if album is None:
raise NotFoundError(f"Album {album_id} not found.")
tracks = await track_repo.list(
artist_id=None,
album_id=album_id,
q=None,
sort_by="title",
order="asc",
limit=limit,
offset=offset,
)
total = await track_repo.count(artist_id=None, album_id=album_id, q=None)
artist_ids = list({t.artist_id for t in tracks})
artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums = {album.id: album}
items = await _build_track_out(tracks, artists, albums)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{album_id}/cover")
async def get_album_cover(album_id: uuid.UUID) -> Any: ...
async def get_album_cover(album_id: uuid.UUID, _: CurrentUser) -> Any: ...
+105 -6
View File
@@ -3,26 +3,125 @@
import uuid
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Query
from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, TrackRepoDep
from app.api.schemas.album import AlbumOut
from app.api.schemas.artist import ArtistOut
from app.api.schemas.pagination import PagedResponse
from app.api.schemas.track import TrackOut
from app.api.v1.albums import _build_album_out
from app.api.v1.tracks import _build_track_out
from app.domain.errors import NotFoundError
router = APIRouter(prefix="/artists", tags=["artists"])
@router.get("")
async def list_artists() -> Any: ...
async def list_artists(
artist_repo: ArtistRepoDep,
_: CurrentUser,
q: str | None = None,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[ArtistOut]:
artists = await artist_repo.list(q=q, limit=limit, offset=offset)
total = await artist_repo.count(q=q)
items = []
for a in artists:
album_cnt = await artist_repo.album_count(a.id)
track_cnt = await artist_repo.track_count(a.id)
items.append(
ArtistOut(
id=a.id,
name=a.name,
album_count=album_cnt,
track_count=track_cnt,
created_at=a.created_at,
)
)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{artist_id}")
async def get_artist(artist_id: uuid.UUID) -> Any: ...
async def get_artist(
artist_id: uuid.UUID,
artist_repo: ArtistRepoDep,
_: CurrentUser,
) -> ArtistOut:
artist = await artist_repo.get_by_id(artist_id)
if artist is None:
raise NotFoundError(f"Artist {artist_id} not found.")
album_cnt = await artist_repo.album_count(artist_id)
track_cnt = await artist_repo.track_count(artist_id)
return ArtistOut(
id=artist.id,
name=artist.name,
album_count=album_cnt,
track_count=track_cnt,
created_at=artist.created_at,
)
@router.get("/{artist_id}/albums")
async def get_artist_albums(artist_id: uuid.UUID) -> Any: ...
async def get_artist_albums(
artist_id: uuid.UUID,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[AlbumOut]:
artist = await artist_repo.get_by_id(artist_id)
if artist is None:
raise NotFoundError(f"Artist {artist_id} not found.")
albums = await album_repo.list(artist_id=artist_id, q=None, limit=limit, offset=offset)
total = await album_repo.count(artist_id=artist_id, q=None)
artists_map = {artist.id: artist}
track_counts = await album_repo.track_count_many([a.id for a in albums])
items = await _build_album_out(albums, artists_map, track_counts)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{artist_id}/tracks")
async def get_artist_tracks(artist_id: uuid.UUID) -> Any: ...
async def get_artist_tracks(
artist_id: uuid.UUID,
artist_repo: ArtistRepoDep,
track_repo: TrackRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[TrackOut]:
artist = await artist_repo.get_by_id(artist_id)
if artist is None:
raise NotFoundError(f"Artist {artist_id} not found.")
tracks = await track_repo.list(
artist_id=artist_id,
album_id=None,
q=None,
sort_by="title",
order="asc",
limit=limit,
offset=offset,
)
total = await track_repo.count(artist_id=artist_id, album_id=None, q=None)
album_ids = list({t.album_id for t in tracks if t.album_id is not None})
artists_map = {artist.id: artist}
albums_map = {a.id: a for a in await album_repo.get_many(album_ids)}
items = await _build_track_out(tracks, artists_map, albums_map)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{artist_id}/similar")
async def get_similar_artists(artist_id: uuid.UUID) -> Any: ...
async def get_similar_artists(artist_id: uuid.UUID, _: CurrentUser) -> Any: ...
+44 -7
View File
@@ -1,15 +1,52 @@
"""Playback history endpoints."""
from typing import Any
from fastapi import APIRouter, Query, Response
from fastapi import APIRouter
from app.api.deps import CurrentUser, HistoryRepoDep, TrackRepoDep
from app.api.schemas.history import HistoryIn, HistoryOut
from app.api.schemas.pagination import PagedResponse
from app.domain.errors import NotFoundError
router = APIRouter(prefix="/history", tags=["history"])
@router.post("", status_code=204)
async def record_history(
body: HistoryIn,
history_repo: HistoryRepoDep,
track_repo: TrackRepoDep,
user: CurrentUser,
) -> Response:
track = await track_repo.get_by_id(body.track_id)
if track is None:
raise NotFoundError(f"Track {body.track_id} not found.")
await history_repo.add(
user_id=user.id,
track_id=body.track_id,
played_at=body.played_at,
play_duration_seconds=body.play_duration_seconds,
completed=body.completed,
)
return Response(status_code=204)
@router.get("")
async def get_history() -> Any: ...
@router.post("")
async def record_history() -> Any: ...
async def get_history(
history_repo: HistoryRepoDep,
user: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[HistoryOut]:
entries = await history_repo.list(user_id=user.id, limit=limit, offset=offset)
total = await history_repo.count(user_id=user.id)
items = [
HistoryOut(
id=e.id,
track_id=e.track_id,
played_at=e.played_at,
play_duration_seconds=e.play_duration_seconds,
completed=e.completed,
)
for e in entries
]
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
+50 -6
View File
@@ -1,19 +1,63 @@
"""Like endpoints. Likes are an append-only event-log — never updated in place."""
from typing import Any
import uuid
from fastapi import APIRouter
from fastapi import APIRouter, Query
from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, LikeRepoDep
from app.api.schemas.like import LikeEvent, LikeState
from app.api.schemas.pagination import PagedResponse
from app.api.schemas.track import TrackOut
from app.api.v1.tracks import _build_track_out
router = APIRouter(prefix="/likes", tags=["likes"])
@router.post("", status_code=201)
async def add_like(
body: LikeEvent,
like_repo: LikeRepoDep,
user: CurrentUser,
) -> LikeState:
like = await like_repo.add(user_id=user.id, track_id=body.track_id, value=body.value)
return LikeState(track_id=like.track_id, value=like.value, updated_at=like.created_at)
@router.get("")
async def get_likes() -> Any: ...
async def get_likes(
like_repo: LikeRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
user: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[TrackOut]:
tracks = await like_repo.list_liked_tracks(user_id=user.id, limit=limit, offset=offset)
total = await like_repo.count_liked_tracks(user_id=user.id)
artist_ids = list({t.artist_id for t in tracks})
album_ids = list({t.album_id for t in tracks if t.album_id is not None})
artists_map = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums_map = {a.id: a for a in await album_repo.get_many(album_ids)}
@router.post("")
async def add_like() -> Any: ...
items = await _build_track_out(tracks, artists_map, albums_map)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/state")
async def get_likes_state() -> Any: ...
async def get_likes_state(
like_repo: LikeRepoDep,
user: CurrentUser,
track_ids: str = Query(default=""),
) -> list[LikeState]:
ids: list[uuid.UUID] = []
if track_ids:
try:
ids = [uuid.UUID(tid.strip()) for tid in track_ids.split(",") if tid.strip()]
except ValueError:
return []
likes = await like_repo.get_latest_state(user_id=user.id, track_ids=ids)
return [
LikeState(track_id=lk.track_id, value=lk.value, updated_at=lk.created_at) for lk in likes
]
+156 -15
View File
@@ -3,46 +3,187 @@
import uuid
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Query, Response
from app.api.deps import (
AlbumRepoDep,
ArtistRepoDep,
CurrentUser,
PlaylistRepoDep,
TrackRepoDep,
)
from app.api.schemas.pagination import PagedResponse
from app.api.schemas.playlist import PlaylistAddTrack, PlaylistCreate, PlaylistOut, PlaylistUpdate
from app.api.schemas.track import TrackOut
from app.api.v1.tracks import _build_track_out
from app.domain.entities.playlist import Playlist
from app.domain.errors import NotFoundError, PermissionDeniedError
from app.infrastructure.db.repositories.playlist_repository import SqlAlchemyPlaylistRepository
router = APIRouter(prefix="/playlists", tags=["playlists"])
async def _build_playlist_out(
playlists: list[Playlist], playlist_repo: SqlAlchemyPlaylistRepository
) -> list[PlaylistOut]:
ids = [p.id for p in playlists]
counts = await playlist_repo.track_count_many(ids)
return [
PlaylistOut(
id=p.id,
name=p.name,
description=p.description,
owner_id=p.owner_id,
version=p.version,
track_count=counts.get(p.id, 0),
created_at=p.created_at,
)
for p in playlists
]
@router.get("")
async def list_playlists() -> Any: ...
async def list_playlists(
playlist_repo: PlaylistRepoDep,
user: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[PlaylistOut]:
playlists = await playlist_repo.list(owner_id=user.id, limit=limit, offset=offset)
total = await playlist_repo.count(owner_id=user.id)
items = await _build_playlist_out(playlists, playlist_repo)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.post("")
async def create_playlist() -> Any: ...
@router.post("", status_code=201)
async def create_playlist(
body: PlaylistCreate,
playlist_repo: PlaylistRepoDep,
user: CurrentUser,
) -> PlaylistOut:
playlist = await playlist_repo.add(
name=body.name, description=body.description, owner_id=user.id
)
items = await _build_playlist_out([playlist], playlist_repo)
return items[0]
@router.get("/{playlist_id}")
async def get_playlist(playlist_id: uuid.UUID) -> Any: ...
async def get_playlist(
playlist_id: uuid.UUID,
playlist_repo: PlaylistRepoDep,
_: CurrentUser,
) -> PlaylistOut:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
items = await _build_playlist_out([playlist], playlist_repo)
return items[0]
@router.patch("/{playlist_id}")
async def update_playlist(playlist_id: uuid.UUID) -> Any: ...
async def update_playlist(
playlist_id: uuid.UUID,
body: PlaylistUpdate,
playlist_repo: PlaylistRepoDep,
user: CurrentUser,
) -> PlaylistOut:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
if playlist.owner_id != user.id:
raise PermissionDeniedError("You don't own this playlist.")
updated = await playlist_repo.update(playlist_id, name=body.name, description=body.description)
items = await _build_playlist_out([updated], playlist_repo)
return items[0]
@router.delete("/{playlist_id}")
async def delete_playlist(playlist_id: uuid.UUID) -> Any: ...
@router.delete("/{playlist_id}", status_code=204)
async def delete_playlist(
playlist_id: uuid.UUID,
playlist_repo: PlaylistRepoDep,
user: CurrentUser,
) -> Response:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
if playlist.owner_id != user.id:
raise PermissionDeniedError("You don't own this playlist.")
await playlist_repo.delete(playlist_id)
return Response(status_code=204)
@router.get("/{playlist_id}/tracks")
async def get_playlist_tracks(playlist_id: uuid.UUID) -> Any: ...
async def get_playlist_tracks(
playlist_id: uuid.UUID,
playlist_repo: PlaylistRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[TrackOut]:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
tracks = await playlist_repo.get_tracks(playlist_id, limit=limit, offset=offset)
total = await playlist_repo.get_track_total(playlist_id)
artist_ids = list({t.artist_id for t in tracks})
album_ids = list({t.album_id for t in tracks if t.album_id is not None})
artists_map = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums_map = {a.id: a for a in await album_repo.get_many(album_ids)}
items = await _build_track_out(tracks, artists_map, albums_map)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.post("/{playlist_id}/tracks")
async def add_playlist_tracks(playlist_id: uuid.UUID) -> Any: ...
@router.post("/{playlist_id}/tracks", status_code=204)
async def add_playlist_track(
playlist_id: uuid.UUID,
body: PlaylistAddTrack,
playlist_repo: PlaylistRepoDep,
track_repo: TrackRepoDep,
user: CurrentUser,
) -> Response:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
if playlist.owner_id != user.id:
raise PermissionDeniedError("You don't own this playlist.")
track = await track_repo.get_by_id(body.track_id)
if track is None:
raise NotFoundError(f"Track {body.track_id} not found.")
position = body.position
if position is None:
position = await playlist_repo.max_position(playlist_id) + 1.0
await playlist_repo.add_track(playlist_id, body.track_id, position=position)
return Response(status_code=204)
@router.delete("/{playlist_id}/tracks/{track_id}")
async def remove_playlist_track(playlist_id: uuid.UUID, track_id: uuid.UUID) -> Any: ...
@router.delete("/{playlist_id}/tracks/{track_id}", status_code=204)
async def remove_playlist_track(
playlist_id: uuid.UUID,
track_id: uuid.UUID,
playlist_repo: PlaylistRepoDep,
user: CurrentUser,
) -> Response:
playlist = await playlist_repo.get_by_id(playlist_id)
if playlist is None:
raise NotFoundError(f"Playlist {playlist_id} not found.")
if playlist.owner_id != user.id:
raise PermissionDeniedError("You don't own this playlist.")
await playlist_repo.remove_track(playlist_id, track_id)
return Response(status_code=204)
@router.put("/{playlist_id}/tracks/reorder")
async def reorder_playlist_tracks(playlist_id: uuid.UUID) -> Any: ...
async def reorder_playlist_tracks(playlist_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.get("/{playlist_id}/cover")
async def get_playlist_cover(playlist_id: uuid.UUID) -> Any: ...
async def get_playlist_cover(playlist_id: uuid.UUID, _: CurrentUser) -> Any: ...
+66 -3
View File
@@ -2,14 +2,77 @@
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Query
from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, TrackRepoDep
from app.api.schemas.album import AlbumOut
from app.api.schemas.artist import ArtistOut
from app.api.schemas.search import LibrarySearchResponse
from app.api.schemas.track import TrackOut
from app.api.v1.albums import _build_album_out
from app.api.v1.tracks import _build_track_out
router = APIRouter(prefix="/search", tags=["search"])
@router.get("")
async def search() -> Any: ...
async def search(_: CurrentUser) -> Any: ...
@router.get("/library")
async def search_library() -> Any: ...
async def search_library(
track_repo: TrackRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
q: str = Query(min_length=1),
types: str = Query(default="tracks,albums,artists"),
limit: int = Query(20, ge=1, le=100),
) -> LibrarySearchResponse:
requested = {t.strip() for t in types.split(",")}
tracks_out: list[TrackOut] = []
albums_out: list[AlbumOut] = []
artists_out: list[ArtistOut] = []
if "tracks" in requested:
tracks = await track_repo.list(
artist_id=None,
album_id=None,
q=q,
sort_by="title",
order="asc",
limit=limit,
offset=0,
)
if tracks:
artist_ids = list({t.artist_id for t in tracks})
album_ids = list({t.album_id for t in tracks if t.album_id is not None})
artists_map = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums_map = {a.id: a for a in await album_repo.get_many(album_ids)}
tracks_out = await _build_track_out(tracks, artists_map, albums_map)
if "albums" in requested:
albums = await album_repo.list(artist_id=None, q=q, limit=limit, offset=0)
if albums:
artist_ids = list({a.artist_id for a in albums})
artists_map = {a.id: a for a in await artist_repo.get_many(artist_ids)}
track_counts = await album_repo.track_count_many([a.id for a in albums])
albums_out = await _build_album_out(albums, artists_map, track_counts)
if "artists" in requested:
raw_artists = await artist_repo.list(q=q, limit=limit, offset=0)
for a in raw_artists:
album_cnt = await artist_repo.album_count(a.id)
track_cnt = await artist_repo.track_count(a.id)
artists_out.append(
ArtistOut(
id=a.id,
name=a.name,
album_count=album_cnt,
track_count=track_cnt,
created_at=a.created_at,
)
)
return LibrarySearchResponse(tracks=tracks_out, albums=albums_out, artists=artists_out)
+123 -13
View File
@@ -1,48 +1,158 @@
"""Track endpoints (library CRUD, similarity, optimization, cover, metadata, streaming)."""
"""Track endpoints."""
import uuid
from typing import Any
from fastapi import APIRouter
from fastapi import APIRouter, Query, Response
from app.api.deps import AlbumRepoDep, ArtistRepoDep, CurrentUser, FileStorageDep, TrackRepoDep
from app.api.schemas.pagination import PagedResponse
from app.api.schemas.track import TrackOut, TrackUpdate
from app.domain.entities.album import Album
from app.domain.entities.track import Artist, Track
from app.domain.errors import NotFoundError
router = APIRouter(prefix="/tracks", tags=["tracks"])
async def _build_track_out(
tracks: list[Track],
artists: dict[uuid.UUID, Artist],
albums: dict[uuid.UUID, Album],
) -> list[TrackOut]:
return [
TrackOut(
id=t.id,
title=t.title,
artist_id=t.artist_id,
artist_name=artists[t.artist_id].name if t.artist_id in artists else "Unknown Artist",
album_id=t.album_id,
album_title=albums[t.album_id].title if t.album_id and t.album_id in albums else None,
duration_seconds=t.duration_seconds,
file_format=t.file_format,
file_size=t.file_size,
metadata_status=t.metadata_status,
source=t.source,
created_at=t.created_at,
)
for t in tracks
]
@router.get("")
async def list_tracks() -> Any: ...
async def list_tracks(
track_repo: TrackRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
artist_id: uuid.UUID | None = None,
album_id: uuid.UUID | None = None,
q: str | None = None,
sort_by: str = Query("created_at", pattern="^(title|created_at|artist)$"),
order: str = Query("desc", pattern="^(asc|desc)$"),
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
) -> PagedResponse[TrackOut]:
tracks = await track_repo.list(
artist_id=artist_id,
album_id=album_id,
q=q,
sort_by=sort_by,
order=order,
limit=limit,
offset=offset,
)
total = await track_repo.count(artist_id=artist_id, album_id=album_id, q=q)
artist_ids = list({t.artist_id for t in tracks})
album_ids = list({t.album_id for t in tracks if t.album_id is not None})
artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums = {a.id: a for a in await album_repo.get_many(album_ids)}
items = await _build_track_out(tracks, artists, albums)
return PagedResponse(items=items, total=total, limit=limit, offset=offset)
@router.get("/{track_id}")
async def get_track(track_id: uuid.UUID) -> Any: ...
async def get_track(
track_id: uuid.UUID,
track_repo: TrackRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
) -> TrackOut:
track = await track_repo.get_by_id(track_id)
if track is None:
raise NotFoundError(f"Track {track_id} not found.")
artist_ids = [track.artist_id]
album_ids = [track.album_id] if track.album_id else []
artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums = {a.id: a for a in await album_repo.get_many(album_ids)}
items = await _build_track_out([track], artists, albums)
return items[0]
@router.patch("/{track_id}")
async def update_track(track_id: uuid.UUID) -> Any: ...
async def update_track(
track_id: uuid.UUID,
body: TrackUpdate,
track_repo: TrackRepoDep,
artist_repo: ArtistRepoDep,
album_repo: AlbumRepoDep,
_: CurrentUser,
) -> TrackOut:
track = await track_repo.update(
track_id,
title=body.title,
genre=body.genre,
year=body.year,
)
artist_ids = [track.artist_id]
album_ids = [track.album_id] if track.album_id else []
artists = {a.id: a for a in await artist_repo.get_many(artist_ids)}
albums = {a.id: a for a in await album_repo.get_many(album_ids)}
items = await _build_track_out([track], artists, albums)
return items[0]
@router.delete("/{track_id}")
async def delete_track(track_id: uuid.UUID) -> Any: ...
@router.delete("/{track_id}", status_code=204)
async def delete_track(
track_id: uuid.UUID,
track_repo: TrackRepoDep,
storage: FileStorageDep,
_: CurrentUser,
) -> Response:
track = await track_repo.get_by_id(track_id)
if track is None:
raise NotFoundError(f"Track {track_id} not found.")
await track_repo.delete(track_id)
await storage.delete(track.file_path)
return Response(status_code=204)
@router.get("/{track_id}/similar")
async def get_similar_tracks(track_id: uuid.UUID) -> Any: ...
async def get_similar_tracks(track_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.post("/{track_id}/optimize")
async def optimize_track(track_id: uuid.UUID) -> Any: ...
async def optimize_track(track_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.get("/{track_id}/cover")
async def get_track_cover(track_id: uuid.UUID) -> Any: ...
async def get_track_cover(track_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.post("/{track_id}/metadata/enrich")
async def enrich_metadata(track_id: uuid.UUID) -> Any: ...
async def enrich_metadata(track_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.get("/{track_id}/metadata/matches")
async def get_metadata_matches(track_id: uuid.UUID) -> Any: ...
async def get_metadata_matches(track_id: uuid.UUID, _: CurrentUser) -> Any: ...
@router.put("/{track_id}/metadata")
async def set_metadata(track_id: uuid.UUID) -> Any: ...
async def set_metadata(track_id: uuid.UUID, _: CurrentUser) -> Any: ...