feat: Add search and sorting functionality to sound repository and API
Some checks failed
Backend CI / lint (push) Failing after 4m54s
Backend CI / test (push) Failing after 3m46s

This commit is contained in:
JSC
2025-08-10 15:33:15 +02:00
parent 8544a3ce22
commit aa9a73ac1d
2 changed files with 101 additions and 3 deletions

View File

@@ -10,7 +10,7 @@ from app.core.dependencies import get_current_active_user_flexible
from app.models.credit_action import CreditActionType from app.models.credit_action import CreditActionType
from app.models.sound import Sound from app.models.sound import Sound
from app.models.user import User from app.models.user import User
from app.repositories.sound import SoundRepository from app.repositories.sound import SoundRepository, SoundSortField, SortOrder
from app.services.credit import CreditService, InsufficientCreditsError from app.services.credit import CreditService, InsufficientCreditsError
from app.services.vlc_player import VLCPlayerService, get_vlc_player_service from app.services.vlc_player import VLCPlayerService, get_vlc_player_service
@@ -42,10 +42,37 @@ async def get_sounds(
list[str] | None, list[str] | None,
Query(description="Filter by sound types (e.g., SDB, TTS, EXT)"), Query(description="Filter by sound types (e.g., SDB, TTS, EXT)"),
] = None, ] = None,
search: Annotated[
str | None,
Query(description="Search sounds by name"),
] = None,
sort_by: Annotated[
SoundSortField | None,
Query(description="Sort by field"),
] = None,
sort_order: Annotated[
SortOrder,
Query(description="Sort order (asc or desc)"),
] = SortOrder.ASC,
limit: Annotated[
int | None,
Query(description="Maximum number of results", ge=1, le=1000),
] = None,
offset: Annotated[
int,
Query(description="Number of results to skip", ge=0),
] = 0,
) -> dict[str, list[Sound]]: ) -> dict[str, list[Sound]]:
"""Get all sounds, optionally filtered by types.""" """Get sounds with optional search, filtering, and sorting."""
try: try:
sounds = await sound_repo.get_by_types(types) sounds = await sound_repo.search_and_sort(
search_query=search,
sound_types=types,
sort_by=sort_by,
sort_order=sort_order,
limit=limit,
offset=offset,
)
except Exception as e: except Exception as e:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,

View File

@@ -1,5 +1,6 @@
"""Sound repository for database operations.""" """Sound repository for database operations."""
from enum import Enum
from sqlalchemy import func from sqlalchemy import func
from sqlmodel import col, select from sqlmodel import col, select
from sqlmodel.ext.asyncio.session import AsyncSession from sqlmodel.ext.asyncio.session import AsyncSession
@@ -11,6 +12,26 @@ from app.repositories.base import BaseRepository
logger = get_logger(__name__) logger = get_logger(__name__)
class SoundSortField(str, Enum):
"""Sound sort field enumeration."""
NAME = "name"
FILENAME = "filename"
DURATION = "duration"
SIZE = "size"
TYPE = "type"
PLAY_COUNT = "play_count"
CREATED_AT = "created_at"
UPDATED_AT = "updated_at"
class SortOrder(str, Enum):
"""Sort order enumeration."""
ASC = "asc"
DESC = "desc"
class SoundRepository(BaseRepository[Sound]): class SoundRepository(BaseRepository[Sound]):
"""Repository for sound operations.""" """Repository for sound operations."""
@@ -107,3 +128,53 @@ class SoundRepository(BaseRepository[Sound]):
except Exception: except Exception:
logger.exception("Failed to get sounds by types: %s", sound_types) logger.exception("Failed to get sounds by types: %s", sound_types)
raise raise
async def search_and_sort(
self,
search_query: str | None = None,
sound_types: list[str] | None = None,
sort_by: SoundSortField | None = None,
sort_order: SortOrder = SortOrder.ASC,
limit: int | None = None,
offset: int = 0,
) -> list[Sound]:
"""Search and sort sounds with optional filtering."""
try:
statement = select(Sound)
# Apply type filter
if sound_types:
statement = statement.where(col(Sound.type).in_(sound_types))
# Apply search filter
if search_query and search_query.strip():
search_pattern = f"%{search_query.strip().lower()}%"
statement = statement.where(
func.lower(Sound.name).like(search_pattern)
)
# Apply sorting
if sort_by:
sort_column = getattr(Sound, sort_by.value)
if sort_order == SortOrder.DESC:
statement = statement.order_by(sort_column.desc())
else:
statement = statement.order_by(sort_column.asc())
else:
# Default sorting by name ascending
statement = statement.order_by(Sound.name.asc())
# Apply pagination
if offset > 0:
statement = statement.offset(offset)
if limit is not None:
statement = statement.limit(limit)
result = await self.session.exec(statement)
return list(result.all())
except Exception:
logger.exception(
"Failed to search and sort sounds: query=%s, types=%s, sort_by=%s, sort_order=%s",
search_query, sound_types, sort_by, sort_order
)
raise