191 lines
7.0 KiB
Python
191 lines
7.0 KiB
Python
"""Favorites management API endpoints."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
|
|
from app.core.dependencies import get_current_active_user
|
|
from app.models.user import User
|
|
from app.schemas.common import MessageResponse
|
|
from app.schemas.favorite import (
|
|
FavoriteCountsResponse,
|
|
FavoriteResponse,
|
|
FavoritesListResponse,
|
|
)
|
|
from app.services.favorite import FavoriteService
|
|
|
|
router = APIRouter(prefix="/favorites", tags=["favorites"])
|
|
|
|
|
|
def get_favorite_service() -> FavoriteService:
|
|
"""Get the favorite service."""
|
|
from app.core.database import get_session_factory
|
|
|
|
return FavoriteService(get_session_factory())
|
|
|
|
|
|
@router.get("/", response_model=FavoritesListResponse)
|
|
async def get_user_favorites(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
limit: Annotated[int, Query(ge=1, le=100)] = 50,
|
|
offset: Annotated[int, Query(ge=0)] = 0,
|
|
) -> FavoritesListResponse:
|
|
"""Get all favorites for the current user."""
|
|
favorites = await favorite_service.get_user_favorites(
|
|
current_user.id, limit, offset,
|
|
)
|
|
return FavoritesListResponse(favorites=favorites)
|
|
|
|
|
|
@router.get("/sounds", response_model=FavoritesListResponse)
|
|
async def get_user_sound_favorites(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
limit: Annotated[int, Query(ge=1, le=100)] = 50,
|
|
offset: Annotated[int, Query(ge=0)] = 0,
|
|
) -> FavoritesListResponse:
|
|
"""Get sound favorites for the current user."""
|
|
favorites = await favorite_service.get_user_sound_favorites(
|
|
current_user.id, limit, offset,
|
|
)
|
|
return FavoritesListResponse(favorites=favorites)
|
|
|
|
|
|
@router.get("/playlists", response_model=FavoritesListResponse)
|
|
async def get_user_playlist_favorites(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
limit: Annotated[int, Query(ge=1, le=100)] = 50,
|
|
offset: Annotated[int, Query(ge=0)] = 0,
|
|
) -> FavoritesListResponse:
|
|
"""Get playlist favorites for the current user."""
|
|
favorites = await favorite_service.get_user_playlist_favorites(
|
|
current_user.id, limit, offset,
|
|
)
|
|
return FavoritesListResponse(favorites=favorites)
|
|
|
|
|
|
@router.get("/counts", response_model=FavoriteCountsResponse)
|
|
async def get_favorite_counts(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> FavoriteCountsResponse:
|
|
"""Get favorite counts for the current user."""
|
|
counts = await favorite_service.get_favorite_counts(current_user.id)
|
|
return FavoriteCountsResponse(**counts)
|
|
|
|
|
|
@router.post("/sounds/{sound_id}", response_model=FavoriteResponse)
|
|
async def add_sound_favorite(
|
|
sound_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> FavoriteResponse:
|
|
"""Add a sound to favorites."""
|
|
try:
|
|
favorite = await favorite_service.add_sound_favorite(current_user.id, sound_id)
|
|
return FavoriteResponse.model_validate(favorite)
|
|
except ValueError as e:
|
|
if "not found" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=str(e),
|
|
) from e
|
|
if "already favorited" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(e),
|
|
) from e
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
) from e
|
|
|
|
|
|
@router.post("/playlists/{playlist_id}", response_model=FavoriteResponse)
|
|
async def add_playlist_favorite(
|
|
playlist_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> FavoriteResponse:
|
|
"""Add a playlist to favorites."""
|
|
try:
|
|
favorite = await favorite_service.add_playlist_favorite(
|
|
current_user.id, playlist_id,
|
|
)
|
|
return FavoriteResponse.model_validate(favorite)
|
|
except ValueError as e:
|
|
if "not found" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=str(e),
|
|
) from e
|
|
if "already favorited" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(e),
|
|
) from e
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
) from e
|
|
|
|
|
|
@router.delete("/sounds/{sound_id}", response_model=MessageResponse)
|
|
async def remove_sound_favorite(
|
|
sound_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> MessageResponse:
|
|
"""Remove a sound from favorites."""
|
|
try:
|
|
await favorite_service.remove_sound_favorite(current_user.id, sound_id)
|
|
return MessageResponse(message="Sound removed from favorites")
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=str(e),
|
|
) from e
|
|
|
|
|
|
@router.delete("/playlists/{playlist_id}", response_model=MessageResponse)
|
|
async def remove_playlist_favorite(
|
|
playlist_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> MessageResponse:
|
|
"""Remove a playlist from favorites."""
|
|
try:
|
|
await favorite_service.remove_playlist_favorite(current_user.id, playlist_id)
|
|
return MessageResponse(message="Playlist removed from favorites")
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=str(e),
|
|
) from e
|
|
|
|
|
|
@router.get("/sounds/{sound_id}/check")
|
|
async def check_sound_favorited(
|
|
sound_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> dict[str, bool]:
|
|
"""Check if a sound is favorited by the current user."""
|
|
is_favorited = await favorite_service.is_sound_favorited(current_user.id, sound_id)
|
|
return {"is_favorited": is_favorited}
|
|
|
|
|
|
@router.get("/playlists/{playlist_id}/check")
|
|
async def check_playlist_favorited(
|
|
playlist_id: int,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
favorite_service: Annotated[FavoriteService, Depends(get_favorite_service)],
|
|
) -> dict[str, bool]:
|
|
"""Check if a playlist is favorited by the current user."""
|
|
is_favorited = await favorite_service.is_playlist_favorited(
|
|
current_user.id, playlist_id,
|
|
)
|
|
return {"is_favorited": is_favorited}
|