6b55ff0e81
- Updated tests for listing users to validate pagination and response format. - Changed mock return values to include total count and pagination details. - Refactored user creation mocks for clarity and consistency. - Enhanced assertions to check for presence of pagination fields in responses. - Adjusted test cases for user retrieval and updates to ensure proper handling of user data. - Improved readability by restructuring mock definitions and assertions across various test files.
259 lines
7.4 KiB
Python
259 lines
7.4 KiB
Python
"""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
|