feat: Implement favorites management API; add endpoints for adding, removing, and retrieving favorites for sounds and playlists
feat: Create Favorite model and repository for managing user favorites in the database feat: Add FavoriteService to handle business logic for favorites management feat: Enhance Playlist and Sound response schemas to include favorite indicators and counts refactor: Update API routes to include favorites functionality in playlists and sounds
This commit is contained in:
JSC
252
app/repositories/favorite.py
Normal file
252
app/repositories/favorite.py
Normal file
@@ -0,0 +1,252 @@
|
||||
"""Repository for managing favorites."""
|
||||
|
||||
from sqlmodel import and_, select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from app.core.logging import get_logger
|
||||
from app.models.favorite import Favorite
|
||||
from app.repositories.base import BaseRepository
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class FavoriteRepository(BaseRepository[Favorite]):
|
||||
"""Repository for managing favorites."""
|
||||
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
"""Initialize the favorite repository.
|
||||
|
||||
Args:
|
||||
session: Database session
|
||||
|
||||
"""
|
||||
super().__init__(Favorite, session)
|
||||
|
||||
async def get_user_favorites(
|
||||
self,
|
||||
user_id: int,
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
) -> list[Favorite]:
|
||||
"""Get all favorites for a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
limit: Maximum number of favorites to return
|
||||
offset: Number of favorites to skip
|
||||
|
||||
Returns:
|
||||
List of user favorites
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = (
|
||||
select(Favorite)
|
||||
.where(Favorite.user_id == user_id)
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
.order_by(Favorite.created_at.desc())
|
||||
)
|
||||
result = await self.session.exec(statement)
|
||||
return list(result.all())
|
||||
except Exception:
|
||||
logger.exception("Failed to get favorites for user: %s", user_id)
|
||||
raise
|
||||
|
||||
async def get_user_sound_favorites(
|
||||
self,
|
||||
user_id: int,
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
) -> list[Favorite]:
|
||||
"""Get sound favorites for a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
limit: Maximum number of favorites to return
|
||||
offset: Number of favorites to skip
|
||||
|
||||
Returns:
|
||||
List of user sound favorites
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = (
|
||||
select(Favorite)
|
||||
.where(and_(Favorite.user_id == user_id, Favorite.sound_id.isnot(None)))
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
.order_by(Favorite.created_at.desc())
|
||||
)
|
||||
result = await self.session.exec(statement)
|
||||
return list(result.all())
|
||||
except Exception:
|
||||
logger.exception("Failed to get sound favorites for user: %s", user_id)
|
||||
raise
|
||||
|
||||
async def get_user_playlist_favorites(
|
||||
self,
|
||||
user_id: int,
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
) -> list[Favorite]:
|
||||
"""Get playlist favorites for a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
limit: Maximum number of favorites to return
|
||||
offset: Number of favorites to skip
|
||||
|
||||
Returns:
|
||||
List of user playlist favorites
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = (
|
||||
select(Favorite)
|
||||
.where(
|
||||
and_(Favorite.user_id == user_id, Favorite.playlist_id.isnot(None))
|
||||
)
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
.order_by(Favorite.created_at.desc())
|
||||
)
|
||||
result = await self.session.exec(statement)
|
||||
return list(result.all())
|
||||
except Exception:
|
||||
logger.exception("Failed to get playlist favorites for user: %s", user_id)
|
||||
raise
|
||||
|
||||
async def get_by_user_and_sound(
|
||||
self, user_id: int, sound_id: int
|
||||
) -> Favorite | None:
|
||||
"""Get a favorite by user and sound.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
sound_id: The sound ID
|
||||
|
||||
Returns:
|
||||
The favorite if found, None otherwise
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = select(Favorite).where(
|
||||
and_(Favorite.user_id == user_id, Favorite.sound_id == sound_id)
|
||||
)
|
||||
result = await self.session.exec(statement)
|
||||
return result.first()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to get favorite for user %s and sound %s", user_id, sound_id
|
||||
)
|
||||
raise
|
||||
|
||||
async def get_by_user_and_playlist(
|
||||
self, user_id: int, playlist_id: int
|
||||
) -> Favorite | None:
|
||||
"""Get a favorite by user and playlist.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
playlist_id: The playlist ID
|
||||
|
||||
Returns:
|
||||
The favorite if found, None otherwise
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = select(Favorite).where(
|
||||
and_(Favorite.user_id == user_id, Favorite.playlist_id == playlist_id)
|
||||
)
|
||||
result = await self.session.exec(statement)
|
||||
return result.first()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to get favorite for user %s and playlist %s",
|
||||
user_id,
|
||||
playlist_id,
|
||||
)
|
||||
raise
|
||||
|
||||
async def is_sound_favorited(self, user_id: int, sound_id: int) -> bool:
|
||||
"""Check if a sound is favorited by a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
sound_id: The sound ID
|
||||
|
||||
Returns:
|
||||
True if the sound is favorited, False otherwise
|
||||
|
||||
"""
|
||||
favorite = await self.get_by_user_and_sound(user_id, sound_id)
|
||||
return favorite is not None
|
||||
|
||||
async def is_playlist_favorited(self, user_id: int, playlist_id: int) -> bool:
|
||||
"""Check if a playlist is favorited by a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
playlist_id: The playlist ID
|
||||
|
||||
Returns:
|
||||
True if the playlist is favorited, False otherwise
|
||||
|
||||
"""
|
||||
favorite = await self.get_by_user_and_playlist(user_id, playlist_id)
|
||||
return favorite is not None
|
||||
|
||||
async def count_user_favorites(self, user_id: int) -> int:
|
||||
"""Count total favorites for a user.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
|
||||
Returns:
|
||||
Total number of favorites
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = select(Favorite).where(Favorite.user_id == user_id)
|
||||
result = await self.session.exec(statement)
|
||||
return len(list(result.all()))
|
||||
except Exception:
|
||||
logger.exception("Failed to count favorites for user: %s", user_id)
|
||||
raise
|
||||
|
||||
async def count_sound_favorites(self, sound_id: int) -> int:
|
||||
"""Count how many users have favorited a sound.
|
||||
|
||||
Args:
|
||||
sound_id: The sound ID
|
||||
|
||||
Returns:
|
||||
Number of users who favorited this sound
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = select(Favorite).where(Favorite.sound_id == sound_id)
|
||||
result = await self.session.exec(statement)
|
||||
return len(list(result.all()))
|
||||
except Exception:
|
||||
logger.exception("Failed to count favorites for sound: %s", sound_id)
|
||||
raise
|
||||
|
||||
async def count_playlist_favorites(self, playlist_id: int) -> int:
|
||||
"""Count how many users have favorited a playlist.
|
||||
|
||||
Args:
|
||||
playlist_id: The playlist ID
|
||||
|
||||
Returns:
|
||||
Number of users who favorited this playlist
|
||||
|
||||
"""
|
||||
try:
|
||||
statement = select(Favorite).where(Favorite.playlist_id == playlist_id)
|
||||
result = await self.session.exec(statement)
|
||||
return len(list(result.all()))
|
||||
except Exception:
|
||||
logger.exception("Failed to count favorites for playlist: %s", playlist_id)
|
||||
raise
|
||||
Reference in New Issue
Block a user