from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.orm import Session from app.api.deps import get_db, get_current_active_superuser from app.models.user import User from app.schemas.song import Song, SongCreate, SongUpdate from app.services.song import ( create_song, delete_song, get_song, get_songs, update_song, ) from app.services.artist import get_artist from app.services.album import get_album router = APIRouter() @router.get("/", response_model=List[Song]) def read_songs( db: Session = Depends(get_db), skip: int = 0, limit: int = 100, artist_id: Optional[int] = Query(None, description="Filter by artist ID"), album_id: Optional[int] = Query(None, description="Filter by album ID"), search: Optional[str] = Query(None, description="Search by song title"), ) -> Any: """ Retrieve songs. """ songs = get_songs( db, skip=skip, limit=limit, artist_id=artist_id, album_id=album_id, search=search ) return songs @router.post("/", response_model=Song) def create_new_song( *, db: Session = Depends(get_db), song_in: SongCreate, current_user: User = Depends(get_current_active_superuser), ) -> Any: """ Create new song. Only superusers can create songs. """ # Check if artist exists artist = get_artist(db, artist_id=song_in.artist_id) if not artist: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Artist not found", ) # Check if album exists if album_id is provided if song_in.album_id: album = get_album(db, album_id=song_in.album_id) if not album: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Album not found", ) # Check if album belongs to the artist if album.artist_id != song_in.artist_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Album doesn't belong to the specified artist", ) song = create_song(db, obj_in=song_in) return song @router.get("/{song_id}", response_model=Song) def read_song( *, db: Session = Depends(get_db), song_id: int, ) -> Any: """ Get song by ID. """ song = get_song(db, song_id=song_id) if not song: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Song not found", ) return song @router.put("/{song_id}", response_model=Song) def update_song_endpoint( *, db: Session = Depends(get_db), song_id: int, song_in: SongUpdate, current_user: User = Depends(get_current_active_superuser), ) -> Any: """ Update a song. Only superusers can update songs. """ song = get_song(db, song_id=song_id) if not song: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Song not found", ) # Check if updating artist_id and if artist exists if song_in.artist_id is not None and song_in.artist_id != song.artist_id: artist = get_artist(db, artist_id=song_in.artist_id) if not artist: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Artist not found", ) # Check if updating album_id and if album exists if song_in.album_id is not None and song_in.album_id != song.album_id: if song_in.album_id == 0: # Remove from album song_in.album_id = None else: album = get_album(db, album_id=song_in.album_id) if not album: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Album not found", ) # Check if album belongs to the artist artist_id = song_in.artist_id if song_in.artist_id is not None else song.artist_id if album.artist_id != artist_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Album doesn't belong to the song's artist", ) song = update_song(db, db_obj=song, obj_in=song_in) return song @router.delete("/{song_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def delete_song_endpoint( *, db: Session = Depends(get_db), song_id: int, current_user: User = Depends(get_current_active_superuser), ) -> Any: """ Delete a song. Only superusers can delete songs. """ song = get_song(db, song_id=song_id) if not song: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Song not found", ) delete_song(db, id=song_id) return None