feat: Add endpoint to retrieve sounds with optional type filtering and implement corresponding repository method
Some checks failed
Backend CI / lint (push) Successful in 9m41s
Backend CI / test (push) Failing after 1m39s

This commit is contained in:
JSC
2025-08-01 22:03:09 +02:00
parent d2d0240fdb
commit 4bbae4c5d4
3 changed files with 127 additions and 2 deletions

View File

@@ -2,12 +2,13 @@
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlmodel.ext.asyncio.session import AsyncSession
from app.core.database import get_db, get_session_factory
from app.core.dependencies import get_current_active_user_flexible
from app.models.credit_action import CreditActionType
from app.models.sound import Sound
from app.models.user import User
from app.repositories.sound import SoundRepository
from app.services.credit import CreditService, InsufficientCreditsError
@@ -34,6 +35,27 @@ async def get_sound_repository(
return SoundRepository(session)
@router.get("/")
async def get_sounds(
current_user: Annotated[User, Depends(get_current_active_user_flexible)], # noqa: ARG001
sound_repo: Annotated[SoundRepository, Depends(get_sound_repository)],
types: Annotated[
list[str] | None,
Query(description="Filter by sound types (e.g., SDB, TTS, EXT)"),
] = None,
) -> dict[str, list[Sound]]:
"""Get all sounds, optionally filtered by types."""
try:
sounds = await sound_repo.get_by_types(types)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to get sounds: {e!s}",
) from e
else:
return {"sounds": sounds}
# VLC PLAYER
@router.post("/play/{sound_id}")

View File

@@ -1,7 +1,7 @@
"""Sound repository for database operations."""
from sqlalchemy import func
from sqlmodel import select
from sqlmodel import col, select
from sqlmodel.ext.asyncio.session import AsyncSession
from app.core.logging import get_logger
@@ -95,3 +95,15 @@ class SoundRepository(BaseRepository[Sound]):
sound_type,
)
raise
async def get_by_types(self, sound_types: list[str] | None = None) -> list[Sound]:
"""Get sounds by types. If types is None or empty, return all sounds."""
try:
statement = select(Sound)
if sound_types:
statement = statement.where(col(Sound.type).in_(sound_types))
result = await self.session.exec(statement)
return list(result.all())
except Exception:
logger.exception("Failed to get sounds by types: %s", sound_types)
raise